基于LangChain的RAG应用LangServe部署：动态输入改造求助

要实现动态传入查询问题和目标语言，并完成检索+翻译的流程，需要调整代码的输入处理逻辑和链式结构，具体步骤如下：

1. 定义结构化输入模型 #

使用Pydantic定义输入格式，让LangServe能够接收并解析用户传入的查询问题和目标语言两个参数：

from pydantic import BaseModel

class RAGTranslationInput(BaseModel):
    query: str  # 用户的查询问题
    target_language: str  # 翻译的目标语言，比如"Chinese"、"English"

2. 重构RAG+翻译链 #

将静态检索逻辑改为动态接收输入，串联检索文档→提取内容→翻译的完整流程：

from langchain_core.runnables import RunnablePassthrough

# 1. 检索文档：根据输入的query获取相关文档并提取内容
def retrieve_relevant_docs(inputs):
    docs = retriever_llm.get_relevant_documents(inputs["query"])
    return "\n\n".join([doc.page_content for doc in docs])  # 将多个文档内容合并成字符串

# 2. 翻译提示模板：动态使用目标语言和检索到的文档内容
translation_prompt = ChatPromptTemplate.from_template(
    "请将以下文档内容准确翻译成{target_language}:\n\n{document_content}"
)

# 3. 组装完整链：输入→检索文档→生成翻译提示→LLM翻译
rag_translation_chain = (
    RunnablePassthrough.assign(document_content=retrieve_relevant_docs)
    | translation_prompt
    | llm
)

3. 更新LangServe路由配置 #

将重构后的链添加到路由，并指定输入模型：

add_routes(
    app,
    rag_translation_chain,
    path="/custom",
    input_type=RAGTranslationInput  # 指定输入格式
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

测试方法 #

启动服务后，访问http://localhost:8000/docs打开Swagger UI，找到/custom接口：

• 在请求体中输入JSON格式的参数，比如：

  {
    "query": "What does finance accounts contain?",
    "target_language": "Chinese"
  }
  

• 发送请求后，即可获得检索到的文档翻译结果。

内容的提问来源于stack exchange，提问作者P Venkat Sai Subash