1. 多智能体系统开发实战:基于LangGraph与Streamlit的AI应用构建
在当今AI技术快速发展的背景下,大语言模型(LLM)已经展现出强大的能力。然而,单一智能体在处理复杂任务时往往力不从心——它可能需要用户反复引导、缺乏环境感知能力,且对话记忆有限。这正是多智能体(Multi-Agent)系统大显身手的场景。
1.1 多智能体系统的核心优势
多智能体系统通过分工协作,能够实现以下单智能体难以完成的任务:
- 分阶段模型调用:在不同处理环节使用最适合的专用模型
- 动态知识检索:当模型知识不足时自动查询外部知识库
- 自我修正机制:多个智能体相互校验,减少幻觉(hallucination)问题
- 并行处理能力:同时处理多个子任务,显著提升效率
1.2 LangGraph的核心价值
LangGraph作为LangChain生态系统的新成员,采用图计算范式来构建智能体系统,其主要优势包括:
- 可视化协作流程:用节点表示智能体,边表示交互关系,直观展现系统架构
- 灵活的状态管理:通过GraphState统一维护对话上下文、文档数据等共享状态
- 内置协调策略:提供条件路由、循环控制等常用模式,减少样板代码
- 无缝集成LangChain:可直接复用已有的Chain、Tool等组件
2. 环境准备与项目架构
2.1 基础环境配置
建议使用Python 3.9+环境,通过以下命令安装核心依赖:
bash复制pip install langgraph langchain-ollama tavily-python streamlit
关键组件说明:
- langgraph:多智能体协调框架
- langchain-ollama:本地模型集成(Ollama)
- tavily-python:联网搜索API
- streamlit:Web应用界面
2.2 项目目录结构
标准化的项目结构有助于团队协作和维护:
code复制.
├── .streamlit/ # Streamlit配置
│ └── config.toml
├── chains/ # 智能体实现
│ ├── generate.py # 回答生成器
│ ├── models.py # 模型加载器
│ └── summary.py # 关键词提取
├── graph/ # 协作图定义
│ ├── graph.py # 图结构
│ └── graph_state.py # 状态模型
├── upload_files/ # 用户上传文档
├── .env # 环境变量
├── app.py # Streamlit入口
├── main.py # CLI入口
└── requirements.txt # 依赖清单
2.3 模型配置策略
在models.py中实现灵活的模型加载机制:
python复制from langchain_ollama import ChatOllama, OllamaEmbeddings
def load_model(model_name: str) -> ChatOllama:
"""加载对话模型"""
return ChatOllama(
model=model_name,
temperature=0.7, # 控制创造性
top_p=0.9, # 核采样参数
repeat_penalty=1.1 # 减少重复
)
def load_embeddings(model_name: str) -> OllamaEmbeddings:
"""加载嵌入模型"""
return OllamaEmbeddings(
model=model_name,
num_ctx=2048 # 上下文窗口大小
)
提示:对于中文场景,推荐使用
qwen:7b或deepseek-r1:7b等支持中文的模型。GPU用户可通过--gpu参数加速推理。
3. 智能体设计与实现
3.1 关键词提取智能体
summary.py实现搜索查询优化功能:
python复制from langchain.prompts import ChatPromptTemplate
class SummaryChain:
def __init__(self, model_name):
self.llm = load_model(model_name)
self.prompt = ChatPromptTemplate.from_template("""
你是一名专业的关键词提取助手。请从用户问题和对话历史中提取核心关键词,
组合成高效的搜索查询。只需返回查询词,不要直接回答问题。
历史对话:{history}
当前问题:{question}
""")
self.chain = self.prompt | self.llm
def invoke(self, input_data):
# 添加查询优化逻辑
result = self.chain.invoke(input_data)
return self._post_process(result.content)
def _post_process(self, query: str) -> str:
"""后处理:移除引号等特殊字符"""
return query.strip('"').replace('"', '')
3.2 回答生成智能体
generate.py实现多源信息整合回答:
python复制class GenerateChain:
def __init__(self, model_name):
self.llm = load_model(model_name)
self.prompt = ChatPromptTemplate.from_template("""
根据以下信息回答问题:
{documents}
对话历史:
{history}
问题:{question}
要求:
1. 优先使用提供的信息
2. 保持专业但友好的语气
3. 不确定时明确说明
""")
self.chain = self.prompt | self.llm
def invoke(self, input_data):
# 实现文档优先级排序
sorted_docs = self._rank_documents(input_data["documents"])
return self.chain.invoke({
**input_data,
"documents": sorted_docs
})
4. 协作图构建与优化
4.1 状态模型设计
graph_state.py定义系统共享状态:
python复制from typing import Literal, Optional
from typing_extensions import TypedDict
class GraphState(TypedDict):
model_name: str
operation_type: Literal["websearch", "file", "chat"]
messages: list # 对话消息
documents: Optional[list] # 处理文档
search_results: Optional[list] # 搜索结果
metadata: dict # 自定义元数据
4.2 核心节点实现
graph.py包含关键业务逻辑:
python复制def route_question(state: GraphState) -> str:
"""智能路由决策"""
last_msg = state["messages"][-1].content
if "搜索" in last_msg or "查" in last_msg:
return "websearch"
elif state.get("documents"):
return "file_process"
else:
return "generate"
def web_search(state: GraphState) -> GraphState:
"""增强版搜索功能"""
search_tool = TavilySearchResults(
k=5, # 结果数量
include_raw_content=True,
search_depth="advanced"
)
results = search_tool.invoke({
"query": state["messages"][-1].content,
"include_domains": ["zh.wikipedia.org"] # 限定中文维基
})
state["search_results"] = results
return state
4.3 图结构配置
python复制def create_graph() -> CompiledStateGraph:
workflow = StateGraph(GraphState)
# 节点注册
workflow.add_node("websearch", web_search)
workflow.add_node("generate", generate)
# 条件路由
workflow.set_conditional_entry_point(
route_question,
{
"websearch": "websearch",
"generate": "generate"
}
)
# 边配置
workflow.add_edge("websearch", "generate")
workflow.add_edge("generate", END)
return workflow.compile(
checkpointer=MemorySaver(),
interrupt_before=["generate"] # 允许人工审核
)
5. 应用界面开发
5.1 Streamlit核心逻辑
app.py实现用户交互界面:
python复制def main():
st.title("多智能体协作系统")
# 会话状态初始化
if "graph" not in st.session_state:
st.session_state.graph = create_graph()
# 聊天界面
user_input = st.chat_input("请输入问题...")
if user_input:
with st.spinner("思考中..."):
state = {
"model_name": "qwen:7b",
"messages": [HumanMessage(user_input)],
"operation_type": "chat"
}
# 流式响应
response = st.write_stream(
stream_graph_updates(
st.session_state.graph,
state,
{"configurable": {"thread_id": "user123"}}
)
)
5.2 高级功能实现
文件处理增强版:
python复制def process_uploaded_file(file):
"""支持多种文档格式"""
if file.type == "application/pdf":
text = extract_pdf_text(file)
elif file.type == "text/plain":
text = file.read().decode()
else:
raise ValueError("不支持的格式")
# 智能分块
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "。", "!", "?"]
)
return splitter.split_text(text)
6. 部署与优化建议
6.1 性能优化技巧
- 模型量化:使用
ollama pull qwen:7b-q4加载4-bit量化模型 - 缓存机制:对频繁查询实现Redis缓存
- 异步处理:对耗时操作使用
asyncio - 批处理:合并相似请求减少模型调用
6.2 扩展方向
- 多模态支持:集成图片、音频处理智能体
- 验证机制:添加事实核查智能体
- 领域适配:针对医疗、法律等垂直领域微调
- 移动端适配:开发React Native应用
实测建议:对于中文场景,将Ollama的
num_ctx参数调整为4096可获得更好的长文本处理能力。同时建议为生成式操作设置temperature=0.3-0.7范围内的值,平衡创造性与准确性。
