给大模型包装服务： LangChain-Serve与FastAPI

明天小事

在使用大模型的过程中免不了遇到一个需求：将大模型应用包装成服务。

看一下基本的代码，还是使用了 decorator 模式，并且实现了 serving 与 slackbot 两类装饰者。

装饰者模式类似于 Jave 中的注解，真的很好用，在上一篇中我就提过将 Agent 中的 Tool 抽象成注解的方法。

很多大模型自带 Gradio webUI ，也看到了对应了 Gradio API 。但今天要说的是 LangChain-serve 项目，官方介绍是可以在几分钟内将 LangChain 应用部署到云上。 LangChain 是 LLM 编程的第一大利器，那他对应的Serve项目也当然是首选了。

项目官方标题如下

LangChain Apps on Production with Jina & FastAPI

翻译过来就是

基于 Jina 和 FastAPI 的 LangChain 生产应用

所有说 LangChain-serve 最终也应该是基于 FastAPI 的。

在项目中，最终实现对LLM的包装是基于 Python 的函数装饰器。例如下面：


             #https://github.com/jina-ai/langchain-serve/blob/main/examples/rest/README.md
@serving
defask(input:str)->str:
search=SerpAPIWrapper()
tools=[
Tool(
name="Search",
func=search.run,
description="usefulforwhenyouneedtoanswerquestionsaboutcurrentevents",
prefix="""Answerthefollowingquestionsasbestyoucan,butspeakingasapiratemightspeak.Youhaveaccesstothefollowingtools:"""
suffix="""Begin!Remembertospeakasapiratewhengivingyourfinalanswer.Uselotsof"Args"
Question:{input}
{agent_scratchpad}"""
returnagent_executor.run(input)

而Serving 装饰器（Decorators）定义如下：


             #https://github.com/jina-ai/langchain-serve/blob/main/lcserve/backend/decorators.py
defserving(
_func=None,
websocket:bool=False,
openai_tracing:bool=False,
auth:Callable=None,
defdecorator(func):
@wraps(func)
asyncdefasync_wrapper(*args,**kwargs):
returnawaitfunc(*args,**kwargs)
@wraps(func)
defsync_wrapper(*args,**kwargs):
returnfunc(*args,**kwargs)
ifinspect.iscoroutinefunction(func):
wrapper=async_wrapper
else:
wrapper=sync_wrapper
_args={
'name':func.__name__,
'doc':func.__doc__,
'params':{
'include_ws_callback_handlers':websocket,
'openai_tracing':openai_tracing,
#IfwebsocketisTrue,passthecallbackhandlerstotheclient.
'auth':auth,
ifwebsocket:
wrapper.__ws_serving__=_args
else:
wrapper.__serving__=_args
returnwrapper
if_funcisNone:
returndecorator
else:
returndecorator(_func)

今天在网上找到一篇菜鸟上关于 Python Decorators 的教程，写得非常好，推荐给大家，向菜鸟致敬。
Python 函数装饰器

同时上面代码里也有 slackbot 的装饰器，也一起附上供参考。


             defslackbot(
_func=None,
commands:Dict[str,Callable]=None,
openai_tracing:bool=False,
defdecorator(func):
@wraps(func)
asyncdefasync_wrapper(*args,**kwargs):
returnawaitfunc(*args,**kwargs)
@wraps(func)
defsync_wrapper(*args,**kwargs):
returnfunc(*args,**kwargs)
ifinspect.iscoroutinefunction(func):
wrapper=async_wrapper
else:
wrapper=sync_wrapper
wrapper.__slackbot__={
'name':func.__name__,
'doc':func.__doc__,
'params':{
'commands':commands,
'openai_tracing':openai_tracing,
returnwrapper
if_funcisNone:
returndecorator
else:
returndecorator(_func)

下面给出一个 FastAPI 的例子。


             #https://github.com/tiangolo/fastapi
fromtypingimportUnion
fromfastapiimportFastAPI
app=FastAPI()
@app.get("/")