
目标:理解 LLM API 调用原理,掌握 LCEL 链式写法,能独立编写
plan()类方法。
大语言模型(LLM)的 API 本质上是一个无状态函数:
输入:消息列表(messages)输出:下一条消息(response)
每次调用都是独立的——模型不记得上次说了什么(这正是我们需要记忆系统的原因)。
model | gpt-4o-mini | |
temperature | ||
timeout | ||
use_responses_api | False |
ChatCompletions API(旧/稳定)/v1/chat/completions适用:GPT-3.5, GPT-4, GPT-4o, GPT-4o-mini特点:稳定,LangChain完整支持Responses API(新/推理模型)/v1/responses适用:o1, o3, gpt-5系列推理模型特点:包含 reasoning items(rs_xxx),需要 store=trueLangChain支持尚不完整→用 use_responses_api=False回退
LangChain 提供了一种管道式写法,用| 连接各组件:
chain = prompt | llm | output_parserresult = chain.invoke({"变量名":"值"})
这就像 Unix 管道:
echo "输入"| grep "过滤"| sort | head -3
┌────────────────────┐┌──────────────┐┌───────────────────┐│ChatPromptTemplate│→│ChatOpenAI│→│StrOutputParser││(格式化 prompt)││(调用 API)││(提取纯文本)│└────────────────────┘└──────────────┘└───────────────────┘.invoke({"task":"..."})AIMessage(...)"步骤1: ...\n步骤2..."
# 不用 LCEL(冗长,难以复用)messages =[HumanMessage(content=f"Break down: {task}")]response = llm.invoke(messages)text = response.content# 用 LCEL(简洁,可组合,可流式)chain = prompt | llm |StrOutputParser()text = chain.invoke({"task": task})
LCEL 的隐藏优势:统一的.invoke() /.stream() /.batch() 接口,切换流式输出只需改.stream()。
# demo_llm.py — 完整可运行示例from dotenv import load_dotenvload_dotenv()from langchain_openai importChatOpenAIfrom langchain_core.prompts importChatPromptTemplatefrom langchain_core.output_parsers importStrOutputParser# 1. 创建 LLM 实例llm =ChatOpenAI(model="gpt-4o-mini",temperature=1,timeout=60,use_responses_api=False,)# 2. 定义 Prompt 模板prompt =ChatPromptTemplate.from_messages([("system","你是一个简洁的解释专家,用一句话回答。"),("human","解释:{concept}")])# 3. 构建 LCEL 链chain = prompt | llm |StrOutputParser()# 4. 调用result = chain.invoke({"concept":"TF-IDF"})print(result)# 输出类似:TF-IDF 是衡量词语在文档中重要性的统计方法,词频高且在其他文档中罕见的词权重更高。
ChatPromptTemplate.from_messages([("system","系统提示,定义 AI 的角色和行为"),("human","用户消息,可包含 {变量}"),("ai","可选:预填充的 AI 回复(few-shot)"),("human","可继续添加更多消息"),])
# system:全局指令,优先级最高,LLM 会严格遵守("system","只用中文回答,禁止使用英文。")# human:用户的具体请求("human","What is Python?")# 即使 human 用了英文,system 的指令会让 LLM 用中文回答
# agent.py 第 30-36 行def plan(self, task: str)-> str:"""Generate plan using LLM"""prompt =ChatPromptTemplate.from_messages([("human","Break down the following task into clear, numbered steps:\n\n{task}")])chain:RunnableSequence= prompt | self.llm |StrOutputParser()return chain.invoke({"task": task})
执行流程分解:
task ="写一个 hello world 脚本"↓prompt.format_messages(task=task)→[HumanMessage("Break down... 写一个 hello world 脚本")]↓llm.invoke([HumanMessage(...)])→AIMessage("1. 创建 hello.py\n2. 写入 print 语句\n3. 运行验证")↓StrOutputParser().invoke(AIMessage(...))→"1. 创建 hello.py\n2. 写入 print 语句\n3. 运行验证"
# main.py 中plan = agent.plan(task)# LLM Call #1:纯规划,不调用工具result = agent.execute(task, plan=plan)# plan 传入 execute,避免重复规划
plan 被合并进execute() 的full_input,让执行阶段的 LLM 直接按计划行动。
def reflect(self, task: str, error: str)-> str:prompt =ChatPromptTemplate.from_messages([("human","Task: {task}\n""Error occurred: {error}\n\n""Reflect on what went wrong and suggest how to fix it.")])chain = prompt | self.llm |StrOutputParser()return chain.invoke({"task": task,"error": error})
规律:plan() 和reflect() 用的是同一个模式:prompt|llm|parser,只是 prompt 内容不同。
打开agent.py,找到plan() 方法,修改 prompt:
# 修改前prompt =ChatPromptTemplate.from_messages([("human","Break down the following task into clear, numbered steps:\n\n{task}")])# 修改后(加一个 system 消息)prompt =ChatPromptTemplate.from_messages([("system","你是一个任务规划专家,始终用中文回答,步骤简洁清晰。"),("human","请将以下任务分解为清晰的编号步骤:\n\n{task}")])
运行验证:
uv run python -c "from dotenv import load_dotenv; load_dotenv()from agent import AIAgenta = AIAgent()print(a.plan('在 test 目录下创建 hello.py'))"
题目:在AIAgent 类中添加一个summarize(text:str)->str 方法,调用 LLM 返回三句话的摘要。
def summarize(self, text: str)-> str:"""用 LLM 将长文本总结为三句话"""# 你的代码(提示:用 LCEL 链,参考 plan() 的写法)pass
验收标准:
agent.summarize("...长文本...") 能返回非空字符串题目:实现一个流式输出版的plan(),将chain.invoke() 改为流式打印,让规划内容逐字出现。
def plan_stream(self, task: str):"""流式输出规划步骤"""prompt =ChatPromptTemplate.from_messages([...])chain = prompt | self.llm |StrOutputParser()# 提示:用 chain.stream({"task": task}) 替代 chain.invoke()for chunk in chain.stream({"task": task}):print(chunk, end="", flush=True)print()# 换行
选择题:下面哪个 LCEL 链的写法是正确的?
# Achain = llm | prompt |StrOutputParser()result = chain.invoke({"task":"hello"})# Bchain = prompt | llm |StrOutputParser()result = chain.invoke({"task":"hello"})# Cchain = prompt |StrOutputParser()| llmresult = chain.invoke({"task":"hello"})# Dchain = prompt | llmresult = chain.invoke({"task":"hello"}).content
答案:B 和 D 都可以工作,但 B 更标准(有 Parser)。
- A 错误:prompt 必须在 llm 之前,llm 的输入是 PromptValue,不是 AIMessage
- C 错误:Parser 在 llm 之前,但 Parser 的输入应该是 AIMessage
- D 可以:直接取
.content属性也行,但不如 StrOutputParser 通用
判断题:
ChatOpenAI 和OpenAI 在 langchain_openai 中是同一个类。temperature=0 表示 LLM 每次都给出相同的答案。| 左边组件的输出会成为右边组件的输入。use_responses_api=False 会让请求走/v1/completions 端点。答案:
- 错(完全不同的类,前者是 Chat 模型,后者是 Completion 模型)
- 接近对(确定性更高,但不保证完全一致)
- 对
- 错(走的是
/v1/chat/completions,不是旧的/v1/completions)
from_messages([("system",...),("human","{var}")]) | |
use_responses_api=False |
