LangServe，这一创新性的工具，专为简化LangChain对象及链的部署而生，将它们转化为REST API，从而赋能开发人员构建更为敏捷高效的应用。与FastAPI的深度融合以及Pydantic的精准数据验证，使得LangServe成为构建微服务的理想选择。不仅如此，LangServe还贴心地配备了客户端，无论是Python还是JavaScript开发者，都能轻松调用部署在服务器上的可运行程序，实现了前后端协同开发的新境界。

借助LangServe的API，开发团队只需聚焦于服务端逻辑的实现，FastAPI则负责搭建起对外的接口桥梁。更令人兴奋的是，通过LangServe提供的RemoteRunnable，客户端可以无缝对接服务端资源。值得注意的是，LangServe不仅提供了Python SDK，还支持TypeScript，这意味着前端开发者同样能在JavaScript环境中调用这些服务，开拓了从前端实现复杂逻辑，如流式输出，的可能性。

尽管官方文档中的API封装精良，但在追求高度定制化的场景下，仅凭add_routes难以满足所有需求。为此，LangServe进一步开放了api_handler接口，赋予开发者根据具体业务场景定制功能的自由度。接下来，让我们一同探索如何利用api_handler构建一个定制化模块，以解决RAG（Retrieval-Augmented Generation）应用中常见的挑战。

场景需求分析： 在RAG应用领域，构建提示词模板往往需要嵌入两个关键元素——上下文与用户问题。上下文，由向量检索出的最相关文档片段拼接而成，其token占用量显著，若完全交由客户端处理，无疑会增加不必要的网络负载。理想情况下，上下文的生成与拼接应在服务端完成，直接融入提示词模板，以此减少数据传输，提升整体效率。

解决方案概览： 为了应对上述挑战，我们将运用LangServe的api_handler接口，设计并实现一个定制化模块。该模块的核心任务是在服务端高效地生成上下文，随后将其无缝融合至提示词模板中，避免了客户端与服务端间的大规模数据交换。具体实现细节将在后续代码示例中详细阐述，敬请期待。

通过这一系列优化措施，LangServe不仅简化了LangChain对象的部署与调用流程，更通过开放的api_handler接口，激发了开发者无限的创新潜力，为构建高效、定制化的RAG应用铺平了道路。

Server

def knowledge_qa_handler_runnable(x: Dict[str, Any]):

    user_id = x.get("user_id", '122')

    query = x.get("query", "12")

    max_tokens = x.get("max_token", "5000")

    data = get_prompt_contexts(user_id,query,[],max_tokens=max_tokens)

    template = get_prompt_template("知识库问答","default")

    prompt = ChatPromptTemplate.from_template(template)

    chain = {"context": lambda c: data, "question": RunnablePassthrough()}| prompt | llm

    return APIHandler(chain, path="/simple")

@app.post("/simple/invoke")

async def test_media(request: Request,runnable: Annotated[APIHandler,                                         Depends(knowledge_qa_handler_runnable)],

                                     )->Response:

    return await runnable.invoke(request)

在接口方法@app.post("/simple/invoke")中，参数runnable通过depends中的knowledge_qa_handler_runnable方法来返回APIHandler,并在knowledge_qa_handler_runnable方法中获取上下文，填充提示词模板等逻辑。

在接口返回时，切记需要带上await，我们需要通过 await 关键字等待它完成并获取其结果，否则会报错【TypeError: 'coroutine' object is not iterable】

 client

openai = RemoteRunnable("http://localhost:8000/simple/")

result = openai.invoke({"user_id": '122', "query": '新质生产力的内涵', "max_token": 5000})

print(result)

在客户端，获取到RemoteRunnable后，通过invoke往服务端传递input参数，参数类型要与服务端类型一致，在上例子，服务端接收的是dict,在客户端需要传递的是dict，最后在服务端解析input，根据解析的参数来获取上下文。