源码级别解析 · 源码解析 · 实战应用 · 架构设计
2026-04-21 | 每日技术深度解读
Langroid：Harness LLMs with Multi-Agent Programming
LLM应用开发的新范式
支持生产级部署的完整解决方案
灵活可扩展的架构设计
智能体的基础架构和设计模式
class Agent(AgentConfig):
"""基础智能体类"""
def __init__(self, config: AgentConfig):
self.config = config
self.llm = config.llm
self.vector_store = config.vector_store
self.tools = config.tools
self.message_sys = MessageSystem()
def llm_response(self, message: str) -> str:
"""LLM响应处理"""
llm_message = LLMMessage(
role=Role.USER,
content=message
)
response = self.llm.generate(llm_message)
return response
def handle_message(self, message: str) -> str:
"""消息处理入口"""
# 消息路由和预处理
processed_msg = self.message_sys.process(message)
# LLM响应生成
response = self.llm_response(processed_msg)
return response支持复杂对话场景的智能体
# ChatAgent配置
chat_agent_config = ChatAgentConfig(
llm=OpenAIGPTConfig(
chat_model=OpenAIChatModel.GPT4o,
max_tokens=1000
),
system_message="你是一个专业助手",
use_tools=True,
handle_llm_no_tool="fallback_response"
)
# 创建ChatAgent
agent = ChatAgent(chat_agent_config)
# 使用ChatAgent
response = agent.llm_response("你好，请介绍一下Langroid")
print(response)基于Actor模型的多智能体协作
# 教师-学生智能体协作
teacher_agent = ChatAgent(teacher_config)
student_agent = ChatAgent(student_config)
# 创建任务
teacher_task = Task(
teacher_agent,
name="Teacher",
system_message="提问并给予反馈"
)
student_task = Task(
student_agent,
name="Student",
system_message="回答问题",
single_round=True
)
teacher_task.add_sub_task(student_task)
teacher_task.run("开始数学教学")复杂任务分解和执行的调度系统
灵活的LLM集成系统
# 多LLM配置示例
llm_configs = {
"openai": OpenAIGPTConfig(
chat_model=OpenAIChatModel.GPT4o,
max_tokens=2000
),
"claude": ClaudeConfig(
model="claude-3-opus",
max_tokens=2000
),
"local": LocalLLMConfig(
model="mistral-7b-instruct",
base_url="http://localhost:11434"
)
}
# 统一的LLM接口
class UnifiedLLM:
def __init__(self, config_name: str):
self.config = llm_configs[config_name]
self.llm = self._create_llm()
def _create_llm(self):
if isinstance(self.config, OpenAIGPTConfig):
return OpenAIGPT(self.config)
elif isinstance(self.config, ClaudeConfig):
return ClaudeLLM(self.config)
elif isinstance(self.config, LocalLLMConfig):
return OllamaLLM(self.config)强大的工具调用能力
# 工具定义和使用
from langroid.agent.tool_message import ToolMessage
class CalculatorTool(ToolMessage):
"""计算器工具"""
name = "calculator"
description = "执行数学计算"
def __init__(self, expression: str):
self.expression = expression
def run(self) -> str:
try:
result = eval(self.expression)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
# 工具集成到ChatAgent
tools = [CalculatorTool]
agent_config = ChatAgentConfig(
llm=openai_config,
tools=tools
)
agent = ChatAgent(agent_config)
# LLM现在可以使用calculator工具
response = agent.llm_response("计算 2 + 2 * 3")智能体间通信的核心基础设施
强大的检索增强生成能力
文档理解问答的专业解决方案
# RAG文档对话配置
rag_config = DocChatAgentConfig(
llm=openai_config,
vector_store=chroma_store,
document_path="./documents/",
chunk_size=1000,
chunk_overlap=200
)
# 创建RAG代理
doc_chat_agent = DocChatAgent(rag_config)
# 文档问答
response = doc_chat_agent.llm_response(
"根据提供的文档，什么是Langroid的主要特性？"
)
print(response) # 包含文档引用的详细回答智能化的数据库交互
# SQL代理配置
sql_config = SQLChatAgentConfig(
llm=openai_config,
db_url="postgresql://localhost:5432/mydb",
table_schemas={
"users": "id, name, email, created_at",
"orders": "id, user_id, amount, status"
}
)
# 创建SQL代理
sql_agent = SQLChatAgent(sql_config)
# 自然语言查询
response = sql_agent.llm_response(
"查找过去30天内注册的所有用户及其订单总数"
)
print(response) # 自动生成的SQL查询结果保证系统稳定运行的重要机制
class ErrorHandler:
"""错误处理器"""
def __init__(self):
self.max_retries = 3
self.timeout_seconds = 30
def handle_llm_error(self, error: Exception):
"""LLM错误处理"""
if isinstance(error, OpenAIError):
# API限流或网络错误
time.sleep(2)
return "稍后重试..."
elif isinstance(error, TimeoutError):
# 超时处理
return "请求超时，请简化问题"
else:
# 其他错误
return "抱歉，遇到了一些问题，请重试"
def handle_tool_error(self, tool: ToolMessage, error: Exception):
"""工具错误处理"""
logger.error(f"工具 {tool.name} 执行失败: {error}")
return f"工具 {tool.name} 执行失败: {str(error)}"配置驱动的框架设计
高性能的多智能体系统
# 缓存系统
class CacheSystem:
"""多级缓存系统"""
def __init__(self):
self.memory_cache = {} # 内存缓存
self.redis_cache = RedisCache() # Redis缓存
def get_cache(self, key: str):
"""获取缓存"""
# 先查内存缓存
if key in self.memory_cache:
return self.memory_cache[key]
# 再查Redis缓存
result = self.redis_cache.get(key)
if result:
self.memory_cache[key] = result # 回填内存缓存
return result
def set_cache(self, key: str, value: str, ttl: int = 3600):
"""设置缓存"""
self.memory_cache[key] = value
self.redis_cache.set(key, value, ttl)
# 异步处理
async def async_llm_call(messages: List[LLMMessage]) -> LLMResponse:
"""异步LLM调用"""
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: llm.generate(messages)
)
return response生产环境监控的重要工具
支持多种媒体类型的智能处理
# 多模态文档处理
class MultiModalProcessor:
"""多模态处理器"""
def __init__(self):
self.pdf_parser = PDFParser()
self.image_processor = ImageProcessor()
self.text_extractor = TextExtractor()
def process_document(self, file_path: str):
"""处理多种格式文档"""
if file_path.endswith('.pdf'):
return self.pdf_parser.parse(file_path)
elif file_path.endswith(('.png', '.jpg', '.jpeg')):
return self.image_processor.process(file_path)
else:
return self.text_extractor.extract(file_path)
def multimodal_llm_call(self, content: dict):
"""多模态LLM调用"""
# 构建多模态消息
messages = []
if 'text' in content:
messages.append(LLMMessage(role=Role.USER, content=content['text']))
if 'image' in content:
messages.append(LLMMessage(role=Role.USER, content=content['image']))
return llm.generate(messages)安全的多智能体系统架构
多样化的应用场景
企业级部署最佳实践
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install -r requirements.txt
# 复制应用代码
COPY . .
# 环境变量
ENV LANGROID_CONFIG=/app/config/production.json
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["python", "app.py"]企业级性能指标
| 功能 | Langroid | LangChain | CrewAI | AutoGen |
|---|---|---|---|---|
| 多智能体协作 | ✅ 优秀 | ✅ 支持 | ✅ 优秀 | ✅ 优秀 |
| 工具调用 | ✅ 原生 | ✅ 丰富 | ✅ 支持 | ✅ 支持 |
| RAG系统 | ✅ 完善 | ✅ 完善 | ✅ 一般 | ✅ 一般 |
| 多模态 | ✅ 支持 | ✅ 支持 | ❌ 不支持 | ✅ 支持 |
| MCP集成 | ✅ 原生 | ❌ 不支持 | ❌ 不支持 | ❌ 不支持 |
| Claude插件 | ✅ 专用 | ❌ 不支持 | ❌ 不支持 | ❌ 不支持 |
开发者友好的框架设计
提高开发效率和质量
解决常见使用问题
框架的持续改进
为什么选择Langroid
简单的三步开始使用
感谢阅读！
访问 https://atcfu.com/ai-articles/langroid-multi-agent-framework/ 回顾本文