下一代 AI 应用架构：LangChain 工作流设计 +70% 内存优化方案

# 创建虚拟环境  conda create -n langchain-env python=3.11  conda activate langchain-env  # 安装核心依赖  pip install langchain langchain-openai duckduckgo-search  pip install langchain-experimental  # 高级功能支持

from langchain_community.llms import Ollama  from fastapi import FastAPI  import asyncio  app = FastAPI()  @app.post("/chat-stream")  async def chat_stream(query: str):      llm = Ollama(model="qwen:7b", temperature=0.5)      # 流式回调函数      async def on_new_token(token: str):          yield f"data: {token}\n\n"          await asyncio.sleep(0.01)      # 执行流式生成      await llm.astream(query, callbacks=[{"on_llm_new_token": on_new_token}])

from langchain.prompts import ChatPromptTemplate, FewShotPromptTemplate  # 1. 定义示例集  examples = [      {"input": "高兴", "output": "笑容满面"},      {"input": "悲伤", "output": "泪流满面"}  ]  # 2. 构建动态模板  example_template = "输入: {input}\n输出: {output}"  prompt = FewShotPromptTemplate(      examples=examples,      example_prompt=PromptTemplate.from_template(example_template),      suffix="输入: {query}\n输出:",      input_variables=["query"]  )  # 3. 注入变量  formatted_prompt = prompt.format(query="愤怒")  print(formatted_prompt)

输入: 高兴  输出: 笑容满面  输入: 悲伤  输出: 泪流满面  输入: 愤怒  输出:

from langchain.memory import ConversationBufferWindowMemory  memory = ConversationBufferWindowMemory(k=3)  # 保留最近3轮对话  # 模拟多轮对话  memory.save_context(      {"input": "我叫Alex"},       {"output": "你好Alex！有什么可以帮你？"}  )  memory.save_context(      {"input": "推荐一本科幻小说"},       {"output": "《三体》非常经典"}  )  # 加载当前记忆  print(memory.load_memory_variables({}))

{    "history": "Human: 推荐一本科幻小说\nAI: 《三体》非常经典"  }

from langchain.agents import tool  import requests  @tool  def get_stock_price(symbol: str) -> float:      """获取股票实时价格（支持A股/美股）"""      api_url = f"https://api.example.com/stock/{symbol}"      response = requests.get(api_url).json()      return response["price"]  # 工具调用测试  print(get_stock_price("AAPL"))  # 输出：176.32

from langchain_community.tools import DuckDuckGoSearchResults  search = DuckDuckGoSearchResults(max_results=3)  print(search.run("特斯拉最新车型"))

from langchain.agents import AgentType, initialize_agent  tools = [get_stock_price, DuckDuckGoSearchResults()]  llm = ChatOpenAI(model="gpt-4o")  # 创建代理  agent = initialize_agent(      tools,      llm,      agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,      verbose=True  )  # 执行复杂查询  agent.run(      "查询特斯拉当前股价，并搜索其最新车型的续航里程"  )

from langchain import RunnableParallel, RunnableBranch  # 定义分支逻辑  def route_query(input):      if "价格" in input:          return "stock_flow"      elif "新闻" in input:          return "news_flow"      return "default_flow"  # 构建并行工作流  workflow = RunnableBranch(      (lambda x: route_query(x) == "stock_flow", stock_analysis_chain),      (lambda x: route_query(x) == "news_flow", news_search_chain),      general_qa_chain  )  # 执行查询  response = workflow.invoke("特斯拉股价多少？")