适用人群：有 Python/后端基础，目标岗位为 AI 应用开发工程师、Agent 开发工程师、智能体全栈工程师 预计学习周期：3-6个月（脱产）/ 6-12个月（在职） 核心理念：每个阶段至少产出一个可展示的 GitHub 项目
python# 1. 安装依赖
# pip install openai python-dotenv
# 2. 配置 API Key（.env 文件）
# OPENAI_API_KEY=sk-xxx
# OPENAI_BASE_URL=https://api.openai.com/v1 # 或其他兼容端点
import os
import json
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL")
)
# ===== 示例1：最基础的对话 =====
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个Python编程助手，回答要简洁，代码要有注释。"},
{"role": "user", "content": "用Python写一个装饰器，计算函数执行时间"}
],
temperature=0.7,
max_tokens=1024
)
print(response.choices[0].message.content)
# ===== 示例2：流式输出 =====
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "解释什么是Transformer的Self-Attention机制"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
bash# 安装 Ollama 后
ollama pull deepseek-r1:7b # 推理模型
ollama pull qwen3:14b # 通义千问3
ollama pull llama4:8b # Meta最新
# Python 调用
# pip install ollama
import ollama
response = ollama.chat(
model="qwen3:14b",
messages=[{"role": "user", "content": "你好，介绍一下你自己"}]
)
print(response["message"]["content"])
python# ===== Zero-shot（零样本） =====
prompt_zero = "将以下句子翻译成英文：今天天气真好。"
# ===== Few-shot（少样本） =====
prompt_few = """将中文翻译成英文：
中文：我喜欢吃火锅。
英文：I like eating hot pot.
中文：明天要下雨。
英文：It will rain tomorrow.
中文：今天天气真好。
英文："""
# ===== Chain-of-Thought（思维链） =====
prompt_cot = """问题：一个水池，进水管3小时注满，出水管5小时排空。两管同时开，几小时注满？
让我们一步步思考：
1. 进水速率：1/3 池/小时
2. 出水速率：1/5 池/小时
3. 净进水速率：1/3 - 1/5 = 5/15 - 3/15 = 2/15 池/小时
4. 注满时间：1 ÷ (2/15) = 15/2 = 7.5 小时
答案：7.5小时
---
问题：一个水池，进水管4小时注满，出水管6小时排空。两管同时开，几小时注满？
让我们一步步思考："""
# ===== ReAct 模式（Reasoning + Acting） =====
react_system_prompt = """你是一个能使用工具的AI助手。请按以下格式回答：
Thought: 我需要思考什么？
Action: 要调用的工具名
Action Input: 工具参数
Observation: 工具返回结果
... (可重复)
Thought: 我现在知道最终答案了
Final Answer: 最终回答
可用工具：
- search(query: str): 搜索互联网
- calculate(expression: str): 数学计算
"""
# ===== 结构化输出（JSON Schema） =====
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "分析以下新闻的情感倾向：今天股市大涨，投资者信心十足"}],
response_format={
"type": "json_schema",
"json_schema": {
"name": "sentiment_analysis",
"schema": {
"type": "object",
"properties": {
"sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
"confidence": {"type": "number", "minimum": 0, "maximum": 1},
"keywords": {"type": "array", "items": {"type": "string"}},
"summary": {"type": "string"}
},
"required": ["sentiment", "confidence", "keywords", "summary"]
}
}
}
)
result = json.loads(response.choices[0].message.content)
print(json.dumps(result, indent=2, ensure_ascii=False))
python# ===== 定义工具函数 =====
def get_weather(city: str, date: str = "今天") -> dict:
"""获取指定城市的天气信息"""
# 模拟天气数据（实际应调用天气API）
weather_data = {
"北京": {"今天": "晴，25°C~32°C，北风3级", "明天": "多云，23°C~30°C"},
"上海": {"今天": "小雨，26°C~31°C", "明天": "阴，24°C~29°C"},
}
return {"city": city, "date": date, "weather": weather_data.get(city, {}).get(date, "无数据")}
# ===== Function Calling 调用 =====
tools = [{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称，如'北京'"},
"date": {"type": "string", "description": "日期，如'今天'或'明天'"}
},
"required": ["city"]
}
}
}]
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "北京明天天气怎么样？适合出门吗？"}],
tools=tools,
tool_choice="auto"
)
# 处理工具调用
tool_calls = response.choices[0].message.tool_calls
if tool_calls:
for tool_call in tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
if func_name == "get_weather":
result = get_weather(**func_args)
print(f"工具调用: {func_name}({func_args})")
print(f"返回结果: {result}")
python# pip install langchain langchain-openai langchain-community chromadb pypdf tiktoken
# pip install sentence-transformers # 本地 embedding
# ==========================================
# Step 1: 文档加载与处理
# ==========================================
from langchain_community.document_loaders import PyPDFLoader, TextLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
import os
# 加载文档
loader = DirectoryLoader(
"./documents/",
glob="**/*.pdf",
loader_cls=PyPDFLoader,
show_progress=True
)
documents = loader.load()
print(f"加载了 {len(documents)} 页文档")
# 智能分块（2026最佳实践：语义分块 + 重叠）
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每块1000字符
chunk_overlap=200, # 重叠200字符，保持上下文连贯
separators=["\n\n", "\n", "。", "！", "？", "；", ".", "!", "?", ";", " ", ""],
length_function=len,
)
chunks = text_splitter.split_documents(documents)
print(f"分割为 {len(chunks)} 个文本块")
# ==========================================
# Step 2: 向量化与存储
# ==========================================
embeddings = OpenAIEmbeddings(model="text-embedding-3-large") # 3072维
# 创建向量数据库（持久化到本地）
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db",
collection_name="my_knowledge_base"
)
print(f"向量数据库已创建，包含 {vectorstore._collection.count()} 条记录")
# ==========================================
# Step 3: 高级检索策略
# ==========================================
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
# 多策略检索器
retriever = vectorstore.as_retriever(
search_type="mmr", # MMR: 最大边际相关性，平衡相关性和多样性
search_kwargs={
"k": 20, # 初检索取20条
"fetch_k": 50, # 从50条中筛选
"lambda_mult": 0.7 # 相关性权重(0=最大多样性, 1=最大相关性)
}
)
# Rerank 重排序（提升精度关键）
model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
compressor = CrossEncoderReranker(model=model, top_n=5)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever
)
# 查询改写（Query Rewriting）
def rewrite_query(original_query: str, chat_history: list = None) -> str:
"""将模糊查询改写为精确的检索查询"""
history_context = ""
if chat_history:
history_context = "\n".join([f"用户: {h['user']}\n助手: {h['assistant']}"
for h in chat_history[-3:]])
response = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": "你是一个查询优化器。将用户的原始问题改写为一个精确的、适合向量检索的查询语句。"
"如果用户使用了代词（如'它'、'这个'），请根据对话历史替换为具体的实体。"
}, {
"role": "user",
"content": f"对话历史:\n{history_context}\n\n原始问题: {original_query}\n\n优化后的查询:"
}],
temperature=0.1
)
return response.choices[0].message.content.strip()
# ==========================================
# Step 4: RAG 生成链
# ==========================================
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
RAG_PROMPT = ChatPromptTemplate.from_messages([
("system", """你是一个基于知识库的问答助手。请严格根据提供的参考资料回答问题。
规则：
1. 如果参考资料中包含答案，请直接引用并注明来源
2. 如果参考资料不包含答案，请明确说"根据现有资料无法回答"
3. 回答要简洁、准确，使用编号列表组织多个要点
4. 在每个要点后标注来源文档名和片段编号
参考资料：
{context}"""),
("human", "{question}")
])
def format_docs(docs):
return "\n\n---\n\n".join([
f"[来源{i+1}] 文档: {doc.metadata.get('source', '未知')}\n{doc.page_content}"
for i, doc in enumerate(docs)
])
# 构建 RAG 链
rag_chain = (
{
"context": compression_retriever | format_docs,
"question": RunnablePassthrough()
}
| RAG_PROMPT
| llm
| StrOutputParser()
)
# ==========================================
# Step 5: 完整问答流程（含评估）
# ==========================================
def ask_with_rag(question: str, chat_history: list = None) -> dict:
"""完整的RAG问答流程"""
# 1. 查询改写
optimized_query = rewrite_query(question, chat_history)
print(f"原始问题: {question}")
print(f"优化查询: {optimized_query}")
# 2. 检索 + 生成
answer = rag_chain.invoke(optimized_query)
# 3. 获取检索到的文档（用于评估）
retrieved_docs = compression_retriever.invoke(optimized_query)
return {
"question": question,
"optimized_query": optimized_query,
"answer": answer,
"sources": [
{
"content": doc.page_content[:200],
"source": doc.metadata.get("source", "未知"),
"page": doc.metadata.get("page", "N/A")
}
for doc in retrieved_docs
]
}
# 使用示例
result = ask_with_rag("公司今年的绩效考核标准是什么？")
print(result["answer"])
print(f"\n来源: {json.dumps(result['sources'], indent=2, ensure_ascii=False)}")
python# pip install ragas datasets
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 忠实度：答案是否忠于检索到的上下文
answer_relevancy, # 答案相关性
context_precision, # 上下文精确率
context_recall, # 上下文召回率
context_entity_recall # 实体召回率
)
from datasets import Dataset
# 构建评估数据集
eval_data = Dataset.from_dict({
"question": [
"什么是RAG？",
"如何优化向量检索的召回率？",
"LangChain和LlamaIndex的区别是什么？"
],
"answer": [
"RAG（Retrieval-Augmented Generation）是一种...",
"优化召回率的方法包括混合检索...",
"LangChain是通用框架，LlamaIndex专注于..."
],
"contexts": [
["RAG是一种结合检索和生成的AI技术..."],
["混合检索结合向量检索和BM25..."],
["LangChain提供全链路工具，LlamaIndex侧重索引..."]
],
"ground_truth": [
"RAG是检索增强生成技术，通过检索外部知识来增强LLM的回答质量",
"可以通过混合检索、查询改写、Rerank重排序来提升召回率",
"LangChain是通用LLM应用框架，LlamaIndex专注于数据索引和检索"
]
})
# 执行评估
scores = evaluate(
eval_data,
metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)
print(scores)
python# ==========================================
# 使用 QLoRA 微调 Qwen2.5-7B（消费级显卡可跑）
# ==========================================
# pip install transformers peft datasets accelerate bitsandbytes torch
import torch
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
TrainingArguments,
Trainer,
BitsAndBytesConfig,
DataCollatorForSeq2Seq
)
from peft import LoraConfig, get_peft_model, TaskType, prepare_model_for_kbit_training
from datasets import Dataset
# ===== 1. 准备微调数据 =====
# 数据格式：指令微调的标准格式
training_data = [
{
"instruction": "你是一个医疗问诊助手。请根据患者描述给出初步建议。",
"input": "我最近总是头痛，特别是下午，有时候还会恶心。",
"output": "根据您的描述（下午头痛+恶心），可能的原因包括：\n1. 紧张性头痛：长时间工作导致颈肩肌肉紧张\n2. 视疲劳：长时间盯屏幕\n3. 血压波动：下午血压可能升高\n\n建议：\n- 每工作1小时休息5-10分钟\n- 检查视力，调整屏幕亮度和距离\n- 监测下午血压\n- 如症状持续加重，请及时就医\n\n⚠️ 以上仅为初步分析，不能替代医生诊断。"
},
# ... 至少准备500-1000条高质量数据
]
def format_instruction(sample):
"""将数据格式化为对话模板"""
return {
"text": f"<|im_start|>system\n{sample['instruction']}<|im_end|>\n"
f"<|im_start|>user\n{sample['input']}<|im_end|>\n"
f"<|im_start|>assistant\n{sample['output']}<|im_end|>"
}
dataset = Dataset.from_list(training_data)
dataset = dataset.map(format_instruction)
# ===== 2. 加载模型（4bit量化） =====
model_name = "Qwen/Qwen2.5-7B-Instruct"
bnb_config = BitsAndBytesConfig(
load_in_4bit=True, # 4bit量化
bnb_4bit_quant_type="nf4", # NF4量化类型
bnb_4bit_compute_dtype=torch.bfloat16, # 计算精度
bnb_4bit_use_double_quant=True # 双重量化，进一步节省显存
)
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=bnb_config,
device_map="auto", # 自动分配到GPU
trust_remote_code=True
)
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token
# 准备模型用于k-bit训练
model = prepare_model_for_kbit_training(model)
# ===== 3. 配置 LoRA =====
lora_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
r=16, # LoRA秩（rank），越大模型能力越强但越慢
lora_alpha=32, # LoRA缩放参数，通常设为2*r
lora_dropout=0.1, # Dropout防止过拟合
target_modules=[ # 目标模块（不同模型名称不同）
"q_proj", "k_proj", "v_proj", "o_proj", # Attention层
"gate_proj", "up_proj", "down_proj" # FFN层
],
bias="none"
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters() # 查看可训练参数占比（通常<1%）
# ===== 4. 训练 =====
training_args = TrainingArguments(
output_dir="./qwen2.5-lora-medical",
num_train_epochs=3,
per_device_train_batch_size=4,
gradient_accumulation_steps=4, # 有效batch_size = 4*4 = 16
learning_rate=2e-4,
warmup_ratio=0.03,
logging_steps=10,
save_steps=200,
save_total_limit=3,
fp16=True,
optim="paged_adamw_8bit",
report_to="none"
)
trainer = Trainer(
model=model,
args=training_args,
train_dataset=dataset,
data_collator=DataCollatorForSeq2Seq(tokenizer=tokenizer, padding=True)
)
# 开始训练（7B模型+QLoRA在RTX4090上约需3-6小时）
trainer.train()
# ===== 5. 保存与推理 =====
model.save_pretrained("./qwen2.5-lora-medical-final")
tokenizer.save_pretrained("./qwen2.5-lora-medical-final")
# 加载并推理
from peft import PeftModel
base_model = AutoModelForCausalLM.from_pretrained(model_name, device_map="auto")
model = PeftModel.from_pretrained(base_model, "./qwen2.5-lora-medical-final")
inputs = tokenizer(
"<|im_start|>system\n你是一个医疗问诊助手。<|im_end|>\n<|im_start|>user\n我最近失眠严重，怎么办？<|im_end|>\n<|im_start|>assistant\n",
return_tensors="pt"
).to(model.device)
outputs = model.generate(**inputs, max_new_tokens=512, temperature=0.7)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))
是否需要引入最新/私有知识？ ├── 是 → 用 RAG └── 否 → 是否需要固化风格/格式？ ├── 是 → 用 微调（LoRA） └── 否 → 用 Prompt Engineering 即可 实际项目中：90% = Prompt + RAG，10% = Prompt + RAG + 微调
python# pip install langchain langchain-openai langchain-community
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
import datetime
import json
# ===== 1. 定义工具 =====
@tool
def get_current_time(format: str = "%Y-%m-%d %H:%M:%S") -> str:
"""获取当前日期和时间。format参数可指定时间格式"""
return datetime.datetime.now().strftime(format)
@tool
def calculate(expression: str) -> str:
"""执行数学计算。expression是数学表达式，如'2+3*4'或'sqrt(16)'"""
import math
try:
# 安全的数学计算（仅允许数学函数）
allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("__")}
result = eval(expression, {"__builtins__": {}}, allowed_names)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
@tool
def save_note(content: str, filename: str = None) -> str:
"""保存内容到本地文本文件。content是要保存的内容"""
if not filename:
filename = f"note_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
with open(filename, "w", encoding="utf-8") as f:
f.write(content)
return f"已保存到 {filename}"
# 搜索工具
search = DuckDuckGoSearchRun()
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
# 工具集
tools = [get_current_time, calculate, save_note, search, wikipedia]
# ===== 2. 创建 Agent =====
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# Agent 系统提示词
system_prompt = """你是一个全能的AI助手，可以使用各种工具来完成任务。
核心原则：
1. 对于需要实时信息的问题，使用搜索工具
2. 对于数学计算，使用计算工具
3. 需要保存重要信息时，使用笔记工具
4. 在回答中说明你使用了哪些工具，以及为什么
5. 如果多个工具能解决问题，选择最直接的那个
6. 如果一个工具的结果不够，可以组合使用多个工具
你的名字是"Jarvis"，请用中文回答。"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 打印思考过程
max_iterations=10, # 最大工具调用次数
handle_parsing_errors=True, # 处理解析错误
return_intermediate_steps=True
)
# ===== 3. 运行 Agent =====
result = agent_executor.invoke({
"input": "搜索2026年诺贝尔物理学奖得主，计算他们的平均年龄，然后把结果保存到文件。"
})
print(f"\n最终答案:\n{result['output']}")
print(f"\n中间步骤:")
for step in result["intermediate_steps"]:
action, observation = step
print(f" 工具: {action.tool}")
print(f" 输入: {action.tool_input}")
print(f" 输出: {str(observation)[:200]}...")
print()
python# pip install langgraph langgraph-checkpoint
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain.tools import tool
import operator
# ===== 1. 定义状态 =====
class AgentState(TypedDict):
messages: Annotated[list, add_messages] # 消息列表（自动追加）
iteration_count: int # 迭代计数（防止死循环）
# ===== 2. 定义节点 =====
llm = ChatOpenAI(model="gpt-4o", temperature=0)
@tool
def weather(city: str) -> str:
"""查询城市天气"""
return f"{city}今天晴，22°C~30°C"
@tool
def web_search(query: str) -> str:
"""网络搜索"""
return f"关于'{query}'的搜索结果：..."
tools = [weather, web_search]
llm_with_tools = llm.bind_tools(tools)
def agent_node(state: AgentState) -> AgentState:
"""Agent节点：调用LLM，决定下一步"""
response = llm_with_tools.invoke(state["messages"])
return {
"messages": [response],
"iteration_count": state["iteration_count"] + 1
}
def tool_node(state: AgentState) -> AgentState:
"""工具节点：执行工具调用"""
messages = state["messages"]
last_message = messages[-1]
tool_messages = []
for tool_call in last_message.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
# 找到对应的工具并执行
for tool in tools:
if tool.name == tool_name:
result = tool.invoke(tool_args)
break
tool_messages.append(
ToolMessage(content=str(result), tool_call_id=tool_call["id"])
)
return {"messages": tool_messages, "iteration_count": state["iteration_count"]}
# ===== 3. 条件路由 =====
def should_continue(state: AgentState) -> Literal["tools", "end"]:
"""判断下一步：继续调用工具 or 结束"""
messages = state["messages"]
last_message = messages[-1]
# 防止无限循环
if state["iteration_count"] >= 15:
return "end"
# 如果有工具调用，继续
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "end"
# ===== 4. 构建图 =====
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)
workflow.add_edge("tools", "agent")
# 编译（带内存检查点）
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
# ===== 5. 运行 =====
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(
{"messages": [HumanMessage(content="北京天气怎么样？用搜索查一下明天的天气")], "iteration_count": 0},
config=config
)
for msg in result["messages"]:
if isinstance(msg, AIMessage):
print(f"AI: {msg.content}")
elif hasattr(msg, "tool_calls") and msg.tool_calls:
print(f"调用工具: {[tc['name'] for tc in msg.tool_calls]}")
pythonfrom langgraph.types import interrupt, Command
class ApprovalState(TypedDict):
content: str # 待审批内容
approved: bool # 是否批准
final_output: str # 最终输出
def generate_content(state: ApprovalState) -> ApprovalState:
"""生成内容"""
return {
"content": f"这是自动生成的邮件草稿:\n\n尊敬的客户：\n感谢您的咨询...\n\n请审批。"
}
def human_approval(state: ApprovalState) -> ApprovalState:
"""人工审批节点 —— 这里会暂停等人类操作"""
# interrupt() 会暂停图执行，返回当前状态给调用者
# 人类审批后通过 Command(resume=...) 继续
approval_result = interrupt({
"message": "请审批以上内容",
"content": state["content"],
"action": "approve_or_reject"
})
# approval_result 是人工传入的值
return {"approved": approval_result.get("approved", False)}
def post_process(state: ApprovalState) -> ApprovalState:
"""后处理"""
if state["approved"]:
return {"final_output": f"✅ 已发送:\n{state['content']}"}
else:
return {"final_output": "❌ 审批未通过，内容已丢弃"}
# 构建图
workflow_approval = StateGraph(ApprovalState)
workflow_approval.add_node("generate", generate_content)
workflow_approval.add_node("approve", human_approval)
workflow_approval.add_node("process", post_process)
workflow_approval.add_edge(START, "generate")
workflow_approval.add_edge("generate", "approve")
workflow_approval.add_edge("approve", "process")
workflow_approval.add_edge("process", END)
app_approval = workflow_approval.compile()
# === 调用方代码 ===
# 第一次调用：运行到审批节点时暂停
config = {"configurable": {"thread_id": "approval-001"}}
# 首次运行 —— 会在 human_approval 节点暂停
try:
result = app_approval.invoke(
{"content": "", "approved": False, "final_output": ""},
config=config
)
except Exception as e:
print(f"暂停等待审批: {e}")
# 人工审批后恢复
# resumed = app_approval.invoke(Command(resume={"approved": True}), config=config)
# print(resumed["final_output"])
pythonfrom typing import Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
# ===== 定义 Agent 角色 =====
RESEARCHER_PROMPT = """你是研究员Agent。职责：
1. 搜索和收集信息
2. 提供数据支撑和事实引用
3. 不发表观点，只提供事实
可用工具: web_search"""
WRITER_PROMPT = """你是写手Agent。职责：
1. 根据研究员提供的信息撰写内容
2. 确保文章结构清晰、语言流畅
3. 可要求研究员补充更多信息"""
CRITIC_PROMPT = """你是审稿Agent。职责：
1. 审核写手的内容质量
2. 检查事实准确性、逻辑一致性
3. 指出问题并要求修改
标准：内容准确、逻辑清晰、无冗余"""
# ===== Supervisor 状态 =====
class SupervisorState(TypedDict):
messages: Annotated[list, add_messages]
next_agent: str
task: str # 原始任务
research_complete: bool
draft_complete: bool
review_complete: bool
iteration: int # 审稿轮次
# ===== Supervisor 节点 =====
def supervisor_node(state: SupervisorState) -> SupervisorState:
"""主管Agent：协调各子Agent的工作"""
# 防止无限审稿循环
if state["iteration"] >= 3:
return {"next_agent": "FINISH", "review_complete": True}
supervisor_prompt = f"""你是项目主管。当前任务：{state['task']}
状态：
- 资料搜集: {'完成' if state['research_complete'] else '未完成'}
- 初稿撰写: {'完成' if state['draft_complete'] else '未完成'}
- 审稿次数: {state['iteration']}
请决定下一步由谁工作。选项：
- RESEARCHER: 搜集更多资料
- WRITER: 撰写/修改文章
- CRITIC: 审稿
- FINISH: 任务完成"""
response = llm.invoke([
HumanMessage(content=supervisor_prompt)
])
return {"next_agent": response.content.strip()}
# ===== 构建 Supervisor 图 =====
supervisor_graph = StateGraph(SupervisorState)
# 添加节点
supervisor_graph.add_node("supervisor", supervisor_node)
supervisor_graph.add_node("researcher", researcher_node)
supervisor_graph.add_node("writer", writer_node)
supervisor_graph.add_node("critic", critic_node)
# 路由函数
def route_to_agent(state: SupervisorState) -> str:
return state["next_agent"].lower()
supervisor_graph.add_edge(START, "supervisor")
supervisor_graph.add_conditional_edges("supervisor", route_to_agent, {
"researcher": "researcher",
"writer": "writer",
"critic": "critic",
"finish": END
})
# 每个子Agent完成后回到supervisor
supervisor_graph.add_edge("researcher", "supervisor")
supervisor_graph.add_edge("writer", "supervisor")
supervisor_graph.add_edge("critic", "supervisor")
app_supervisor = supervisor_graph.compile()
# 运行
result = app_supervisor.invoke({
"messages": [HumanMessage(content="写一篇关于AI Agent发展趋势的博客文章")],
"task": "写一篇关于2026年AI Agent发展趋势的博客文章（约1000字）",
"iteration": 0,
"research_complete": False,
"draft_complete": False,
"review_complete": False,
"next_agent": "supervisor"
})
python# ==========================================
# MCP Server 示例：天气服务
# pip install mcp
# ==========================================
# weather_server.py
import asyncio
import json
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource, ResourceTemplate
# 创建 MCP Server
server = Server("weather-service")
# ===== 注册工具 =====
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="get_weather",
description="获取指定城市的实时天气信息",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位",
"default": "celsius"
}
},
"required": ["city"]
}
),
Tool(
name="get_forecast",
description="获取未来7天天气预报",
inputSchema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
"days": {
"type": "integer",
"minimum": 1,
"maximum": 7,
"default": 3
}
},
"required": ["city"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""处理工具调用"""
if name == "get_weather":
city = arguments["city"]
unit = arguments.get("unit", "celsius")
# 模拟天气API调用
weather_data = {
"北京": {"temp": 25, "humidity": 45, "condition": "晴", "wind": "北风3级"},
"上海": {"temp": 28, "humidity": 70, "condition": "小雨", "wind": "东风2级"},
}
data = weather_data.get(city, {"temp": 22, "humidity": 55, "condition": "多云", "wind": "微风"})
temp = data["temp"] if unit == "celsius" else data["temp"] * 9/5 + 32
return [TextContent(
type="text",
text=f"{city}天气: {data['condition']}\n温度: {temp}°{'C' if unit == 'celsius' else 'F'}\n"
f"湿度: {data['humidity']}%\n风力: {data['wind']}"
)]
elif name == "get_forecast":
city = arguments["city"]
days = arguments.get("days", 3)
# 模拟预报
forecast = "\n".join([
f"第{i+1}天: {'晴' if i%2==0 else '多云'} {22+i}°C"
for i in range(days)
])
return [TextContent(type="text", text=f"{city}未来{days}天预报:\n{forecast}")]
raise ValueError(f"未知工具: {name}")
# ===== 注册资源（可被客户端读取的数据） =====
@server.list_resources()
async def list_resources() -> list[Resource]:
return [
Resource(
uri="weather://cities",
name="支持的城市列表",
description="本服务支持查询的城市",
mimeType="application/json"
)
]
@server.read_resource()
async def read_resource(uri: str) -> str:
if uri == "weather://cities":
return json.dumps(["北京", "上海", "广州", "深圳", "杭州", "成都", "武汉"], ensure_ascii=False)
raise ValueError(f"未知资源: {uri}")
# ===== 启动服务 =====
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationCapabilities(
sampling={},
experimental={},
roots={}
),
notification_options=NotificationOptions()
)
if __name__ == "__main__":
asyncio.run(main())
python# ==========================================
# MCP Client：在 LangChain Agent 中使用 MCP 工具
# ==========================================
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp import MCPToolkit
async def create_mcp_agent():
# 连接 MCP Server
server_params = StdioServerParameters(
command="python",
args=["weather_server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化会话
await session.initialize()
# 获取 MCP 工具
toolkit = MCPToolkit(session=session)
tools = await toolkit.get_tools()
# 创建 Agent
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(
model=ChatOpenAI(model="gpt-4o"),
tools=tools
)
# 运行
result = await agent.ainvoke({
"messages": [HumanMessage(content="北京今天天气怎么样？适合户外运动吗？")]
})
print(result["messages"][-1].content)
asyncio.run(create_mcp_agent())
json// langgraph.json —— 部署配置
{
"dependencies": ["."],
"graphs": {
"weather_agent": "./src/weather_agent.py:app",
"multi_agent": "./src/supervisor.py:app_supervisor"
},
"env": ".env",
"python_version": "3.12"
}
yaml# docker-compose.yml
version: '3.8'
services:
langgraph-api:
image: langgraph-api:latest
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- LANGCHAIN_TRACING_V2=true
- LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
volumes:
- ./chroma_db:/app/chroma_db
- ./models:/app/models
restart: always
python# FastAPI 包装 Agent 为 REST API
# app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.memory import MemorySaver
import uuid
app = FastAPI(title="AI Agent API")
class ChatRequest(BaseModel):
message: str
thread_id: str = None
class ChatResponse(BaseModel):
answer: str
thread_id: str
tool_calls: list = []
# 全局Agent实例
agent_app = create_agent_app() # 你的LangGraph应用
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
thread_id = request.thread_id or str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
try:
result = await agent_app.ainvoke(
{"messages": [HumanMessage(content=request.message)], "iteration_count": 0},
config=config
)
# 提取工具调用
tool_calls = []
for msg in result.get("messages", []):
if hasattr(msg, "tool_calls") and msg.tool_calls:
tool_calls.extend([tc["name"] for tc in msg.tool_calls])
return ChatResponse(
answer=result["messages"][-1].content,
thread_id=thread_id,
tool_calls=tool_calls
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health():
return {"status": "healthy"}
bash# 安装仿真环境（无需硬件）
pip install mujoco gymnasium stable-baselines3 torch
# 安装 LeRobot（HuggingFace机器人框架）
pip install lerobot
# 零基础入门工具：zero2robot（完全浏览器内运行）
git clone https://github.com/kaushikb11/zero2robot.git
cd zero2robot
uv sync
python# ==========================================
# 示例1：经典倒立摆（强化学习入门）
# ==========================================
import gymnasium as gym
import numpy as np
env = gym.make("CartPole-v1", render_mode="human")
for episode in range(5):
state, info = env.reset()
total_reward = 0
done = False
while not done:
# 随机策略（先看看效果）
action = env.action_space.sample()
state, reward, terminated, truncated, info = env.step(action)
total_reward += reward
done = terminated or truncated
print(f"Episode {episode+1}: 总奖励 = {total_reward}")
env.close()
# ==========================================
# 示例2：MuJoCo 机械臂仿真
# ==========================================
import mujoco
import mujoco.viewer
import numpy as np
# 加载一个简单的机械臂模型
xml = """
<mujoco>
<worldbody>
<light pos="0 0 3"/>
<body pos="0 0 0">
<joint type="hinge" axis="0 0 1"/>
<geom type="box" size=".5 .1 .05" rgba=".8 .2 .2 1"/>
<body pos="1 0 0">
<joint type="hinge" axis="0 0 1"/>
<geom type="box" size=".3 .05 .03" rgba=".2 .8 .2 1"/>
</body>
</body>
</worldbody>
</mujoco>
"""
model = mujoco.MjModel.from_xml_string(xml)
data = mujoco.MjData(model)
# 简单控制循环
for i in range(1000):
# 施加正弦控制信号
data.ctrl[0] = np.sin(i * 0.1) # 第一个关节
data.ctrl[1] = np.cos(i * 0.1) # 第二个关节
mujoco.mj_step(model, data)
if i % 100 == 0:
print(f"Step {i}: 关节角度 = {data.qpos.round(3)}")
print("仿真完成！")
基于复旦大学 OpenMOSS Lab 的 EI-Beginner 教程
python# ==========================================
# 双臂机器人 PID 轨迹跟踪
# ==========================================
import numpy as np
class PIDController:
"""PID 控制器"""
def __init__(self, kp, ki, kd, dt):
self.kp = kp
self.ki = ki
self.kd = kd
self.dt = dt
self.prev_error = 0
self.integral = 0
def compute(self, target, current):
error = target - current
self.integral += error * self.dt
derivative = (error - self.prev_error) / self.dt
output = self.kp * error + self.ki * self.integral + self.kd * derivative
self.prev_error = error
return output
# 使用 PID 控制机械臂到达目标位置
def pid_arm_control(target_angles, current_angles, dt=0.01):
"""PID控制双臂到达目标角度"""
pid_joints = [PIDController(kp=10.0, ki=0.1, kd=1.0, dt=dt) for _ in range(6)]
trajectory = [current_angles.copy()]
current = current_angles.copy()
for step in range(500):
for j, pid in enumerate(pid_joints):
torque = pid.compute(target_angles[j], current[j])
# 简化的动力学更新
current[j] += torque * dt
trajectory.append(current.copy())
# 检查是否到达目标
if np.allclose(current, target_angles, atol=0.01):
print(f"到达目标，步数: {step}")
break
return np.array(trajectory)
# 使用
current = np.array([0.0, 0.0, 0.0, 0.0, 0.0, 0.0])
target = np.array([0.5, -0.3, 0.0, 1.0, -0.5, 0.8])
traj = pid_arm_control(target, current)
python# ==========================================
# 从人类示教数据学习策略
# ==========================================
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
# 示教数据集（通常通过遥操作或VR设备采集）
class DemonstrationDataset(Dataset):
def __init__(self, observations, actions):
"""
observations: (N, obs_dim) —— 机械臂关节角度、目标位置等
actions: (N, action_dim) —— 关节力矩或目标角度
"""
self.obs = torch.FloatTensor(observations)
self.actions = torch.FloatTensor(actions)
def __len__(self):
return len(self.obs)
def __getitem__(self, idx):
return self.obs[idx], self.actions[idx]
# 策略网络
class BehaviorCloningPolicy(nn.Module):
def __init__(self, obs_dim, action_dim, hidden_dim=256):
super().__init__()
self.net = nn.Sequential(
nn.Linear(obs_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, action_dim),
)
def forward(self, obs):
return self.net(obs)
# 训练
def train_bc(demos: DemonstrationDataset, epochs=100):
dataloader = DataLoader(demos, batch_size=64, shuffle=True)
policy = BehaviorCloningPolicy(obs_dim=12, action_dim=6)
optimizer = torch.optim.Adam(policy.parameters(), lr=1e-3)
criterion = nn.MSELoss()
for epoch in range(epochs):
total_loss = 0
for obs, action in dataloader:
pred_action = policy(obs)
loss = criterion(pred_action, action)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
if epoch % 10 == 0:
print(f"Epoch {epoch}: Loss = {total_loss/len(dataloader):.6f}")
return policy
# 部署
def deploy_policy(policy, env, num_episodes=5):
"""在仿真环境中部署训练好的策略"""
successes = 0
for ep in range(num_episodes):
obs, _ = env.reset()
done = False
while not done:
with torch.no_grad():
action = policy(torch.FloatTensor(obs)).numpy()
obs, reward, terminated, truncated, _ = env.step(action)
done = terminated or truncated
if terminated:
successes += 1
print(f"成功率: {successes}/{num_episodes} ({successes/num_episodes*100:.1f}%)")
python# ==========================================
# 使用 OpenVLA 进行多模态机器人控制
# ==========================================
# pip install transformers torch pillow
from transformers import AutoModelForVision2Seq, AutoProcessor
from PIL import Image
import torch
# 加载 OpenVLA 模型（开源VLA模型）
model = AutoModelForVision2Seq.from_pretrained(
"openvla/openvla-7b",
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
processor = AutoProcessor.from_pretrained("openvla/openvla-7b")
def vla_predict(image: Image.Image, instruction: str) -> dict:
"""
输入：RGB图像 + 自然语言指令
输出：机械臂动作（7维：xyz位移 + 旋转 + 夹爪开合度）
"""
# 构建提示
prompt = f"In: What action should the robot take to {instruction}?\nOut:"
# 处理输入
inputs = processor(
text=prompt,
images=image,
return_tensors="pt"
).to(model.device, dtype=torch.bfloat16)
# 推理
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=50,
do_sample=False
)
# 解析动作
action_text = processor.decode(outputs[0], skip_special_tokens=True)
# 解析7维动作向量（格式因模型而异）
action = parse_action_from_text(action_text)
return {
"action": action, # [dx, dy, dz, droll, dpitch, dyaw, gripper]
"raw_output": action_text
}
def parse_action_from_text(text: str):
"""从模型输出中解析动作向量"""
import re
# 匹配类似 [0.1, -0.05, 0.02, ...] 的格式
pattern = r'\[([^\]]+)\]'
match = re.search(pattern, text)
if match:
nums = [float(x.strip()) for x in match.group(1).split(',')]
return nums
return [0.0] * 7
# 使用示例（需要在仿真或真机环境）
# image = capture_camera_image() # 获取当前图像
# result = vla_predict(image, "pick up the red cube")
# robot.execute_action(result["action"])
python# ==========================================
# 使用 PPO 训练机械臂抓取策略
# ==========================================
import gymnasium as gym
from stable_baselines3 import PPO
from stable_baselines3.common.env_checker import check_env
from stable_baselines3.common.callbacks import EvalCallback
from stable_baselines3.common.monitor import Monitor
import numpy as np
# 使用标准的机械臂环境
# pip install gymnasium-robotics
import gymnasium_robotics
gymnasium_robotics.register_robotics_envs()
# 创建环境
env = gym.make("FetchReach-v3", render_mode="rgb_array")
check_env(env) # 验证环境接口
# 创建 PPO 模型
model = PPO(
policy="MultiInputPolicy", # 处理字典观测空间
env=env,
learning_rate=3e-4,
n_steps=2048,
batch_size=64,
n_epochs=10,
gamma=0.99,
gae_lambda=0.95,
clip_range=0.2,
ent_coef=0.01, # 熵系数，鼓励探索
verbose=1,
tensorboard_log="./ppo_fetch_tensorboard/"
)
# 评估回调
eval_env = Monitor(gym.make("FetchReach-v3"))
eval_callback = EvalCallback(
eval_env,
best_model_save_path="./ppo_fetch_best/",
log_path="./ppo_fetch_logs/",
eval_freq=10000,
n_eval_episodes=10,
deterministic=True
)
# 训练（约需 1M steps，RTX4090 约 2-4 小时）
model.learn(
total_timesteps=1_000_000,
callback=eval_callback,
progress_bar=True
)
model.save("ppo_fetch_reach_final")
# ===== 推理 =====
model = PPO.load("ppo_fetch_reach_final")
obs, _ = env.reset()
for _ in range(100):
action, _ = model.predict(obs, deterministic=True)
obs, reward, terminated, truncated, _ = env.step(action)
if terminated or truncated:
obs, _ = env.reset()
python# ==========================================
# 域随机化 —— Sim-to-Real 的核心技术
# ==========================================
import numpy as np
class DomainRandomizedEnv:
"""
在训练时随机化物理参数，让策略学会应对真实世界的各种变化
"""
def __init__(self, base_env):
self.env = base_env
# 随机化范围
self.randomization_ranges = {
"gravity": (-9.5, -10.5), # 重力加速度
"friction": (0.5, 1.5), # 摩擦系数
"mass_scale": (0.8, 1.2), # 物体质量缩放
"joint_damping": (0.8, 1.2), # 关节阻尼
"obs_noise": (0.0, 0.01), # 观测噪声
"action_noise": (0.0, 0.02), # 执行噪声
}
def randomize(self):
"""随机化环境参数"""
params = {}
for name, (low, high) in self.randomization_ranges.items():
params[name] = np.random.uniform(low, high)
# 应用到仿真环境
self.env.model.opt.gravity[2] = params["gravity"]
# ... 实际实现需要根据具体仿真器API设置
return params
def reset(self):
self.randomize()
return self.env.reset()
def step(self, action):
# 添加执行噪声
noisy_action = action + np.random.normal(0, self.randomization_ranges["action_noise"][1])
obs, reward, term, trunc, info = self.env.step(noisy_action)
# 添加观测噪声
obs += np.random.normal(0, self.randomization_ranges["obs_noise"][1])
return obs, reward, term, trunc, info
# 使用域随机化训练
base_env = gym.make("FetchPickAndPlace-v3")
dr_env = DomainRandomizedEnv(base_env)
model = PPO("MultiInputPolicy", dr_env, verbose=1)
model.learn(total_timesteps=2_000_000)
model.save("ppo_domain_randomized")
| 阶段 | 资源 | 说明 |
|---|---|---|
| 零基础 | zero2robot | 43章免费课程，浏览器内运行 |
| 入门 | EI-Beginner | 复旦OpenMOSS Lab 6阶段教程 |
| 系统学习 | Xbotics Embodied Guide | 中文社区：综述→路线→仿真→开源实物 |
| 资源大全 | Awesome Physical AI | 仿真/RL/IL/VLA/基准/硬件 全收集 |
| 学术入门 | Embodied AI Reading Station | 156篇论文笔记+11大主题阅读路线 |
| 框架 | LeRobot | HuggingFace机器人框架（数据→训练→部署） |
| 仿真 | MuJoCo + Isaac Lab | 两大主流仿真平台 |
| 书籍 | The Principles and Practice of Embodied Intelligence (2026.8) | 系统综述：感知/导航/操作/规划/协作 |
架构： ┌──────────────────────────────────────┐ │ 用户界面 │ │ (Streamlit / Gradio) │ └──────────────┬───────────────────────┘ │ ┌──────────────▼───────────────────────┐ │ Supervisor Agent │ │ (任务分类 → 路由到子Agent) │ └───┬──────────┬──────────┬───────────┘ │ │ │ ┌───▼───┐ ┌───▼───┐ ┌───▼────┐ │ RAG │ │ 搜索 │ │ 分析 │ │ Agent │ │ Agent │ │ Agent │ │(文档 │ │(网络 │ │(数据 │ │ 问答) │ │ 搜索) │ │ 分析) │ └───┬───┘ └───┬───┘ └───┬────┘ │ │ │ ┌───▼──────────▼──────────▼───────────┐ │ 共享工具层 │ │ (向量搜索 / Web搜索 / Python执行) │ └──────────────────────────────────────┘
python# 综合项目骨架：月度复盘Agent
class MonthlyReviewAgent:
"""自动执行月度复盘：读数据→分析→生成报告→发邮件"""
def __init__(self):
self.agents = {
"data_collector": self._create_data_agent(),
"analyzer": self._create_analysis_agent(),
"writer": self._create_writer_agent(),
"reviewer": self._create_reviewer_agent()
}
self.supervisor = self._create_supervisor()
def run_monthly_review(self, month: str):
# 1. 数据收集Agent：从数据库/日志读取
raw_data = self.agents["data_collector"].invoke({
"task": f"收集{month}的业务数据：销售额、用户增长、关键事件"
})
# 2. 分析Agent：数据洞察
analysis = self.agents["analyzer"].invoke({
"task": "分析业务数据，找出趋势、异常、亮点和问题",
"data": raw_data
})
# 3. 写作Agent：生成报告
draft = self.agents["writer"].invoke({
"task": "根据分析结果撰写月度复盘报告",
"analysis": analysis
})
# 4. 审稿Agent：质量把关
final = self.agents["reviewer"].invoke({
"task": "审核报告质量，修正事实错误",
"draft": draft
})
# 5. Human-in-the-Loop：最终审批后发送
return final
| 问题 | 考察点 | 准备要点 |
|---|---|---|
| RAG全链路原理？ | 系统理解 | 文档处理→分块→向量化→检索→重排序→生成，每个环节的优化点 |
| 如何提升召回率？ | 优化能力 | 混合检索、查询改写、HyDE、Rerank、多路召回融合 |
| 如何处理多轮对话中的RAG？ | 工程经验 | 查询改写融合历史、滑动窗口、对话状态管理 |
| 向量数据库选型？ | 架构决策 | Milvus(生产) vs Chroma(原型) vs FAISS(轻量) vs Qdrant(过滤强) |
| RAG评估怎么做？ | 质量保障 | Ragas指标、人工评估、A/B测试、回归测试 |
| 问题 | 考察点 | 准备要点 |
|---|---|---|
| ReAct模式原理？ | 核心机制 | 推理+行动的交替循环，与传统链式调用的区别 |
| 如何防止Agent死循环？ | 工程能力 | max_iterations、收敛检测、Human-in-the-Loop |
| Multi-Agent架构设计？ | 系统设计 | Supervisor、Swarm、层次化、何时用哪种模式 |
| Agent如何保证安全性？ | 安全意识 | 工具权限最小化、提示注入防护、人工审批关键操作 |
| MCP协议是什么？ | 前沿技术 | AI的USB接口、Tools/Resources/Prompts三大能力、传输方式 |
最后的话：2026年是Agent元年。建议优先深耕方向A+B（RAG + Agent），这两项技能组合是当前市场需求最大、薪资溢价最高、且纯软件方向最适合软件工程师转型的路径。方向C（具身智能）需要硬件/仿真背景，适合有机电/机器人基础的同学。记住：项目作品 > 证书，原理理解 > API调用，工程能力 > Demo能力。
本文作者:张京
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！