OpenJiuwen Agent-Core 技术教程

概述

agent-core 是构建 AI Agent 应用的核心 Python SDK，提供高性能运行时引擎和全面的开发工具包，用于创建智能对话系统。

技术栈

• 编程语言: Python 3.11+ (支持 3.11, 3.12, 3.13)
• 核心框架: 自定义 Agent 框架，支持 async/await
• 包管理器: uv (现代 Python 包管理器)
• 关键依赖:
  • OpenAI, Transformers, TikToken (LLM 集成)
  • FastMCP (模型上下文协议服务器)
  • SQLAlchemy (ORM)
  • PyMilvus (向量数据库)
  • aiohttp, requests (HTTP 客户端)

核心特性

1. Agent 类型

ReActAgent

遵循 Reasoning + Action 范式，通过迭代思考-行动-观察循环执行任务。

from openjiuwen import ReActAgent

agent = ReActAgent(
    name="my_agent",
    llm_config={
        "model": "gpt-4",
        "temperature": 0.7
    }
)

result = await agent.run("分析这段代码的性能瓶颈")

WorkflowAgent

流程自动化 Agent，支持预定义工作流和灵活任务切换。

from openjiuwen import WorkflowAgent

agent = WorkflowAgent(
    workflow_definition="path/to/workflow.yaml"
)

result = await agent.execute()

2. 高性能执行引擎

• 异步并行图执行器: 支持组件并发执行
• 流式处理: 实时输出处理结果
• 状态中断与恢复: 支持工作流暂停和恢复

from openjiuwen.core.runner import AsyncGraphRunner

runner = AsyncGraphRunner(
    max_concurrency=10,
    enable_streaming=True
)

async for result in runner.run_stream(workflow):
    print(result)

3. 多工作流管理

from openjiuwen.core.workflow import WorkflowManager

manager = WorkflowManager()

# 注册多个工作流
manager.register_workflow("analysis", analysis_workflow)
manager.register_workflow("generation", generation_workflow)

# 自由切换工作流
await manager.switch_to("analysis")
result = await manager.execute()

4. Prompt 开发工具

from openjiuwen.dev_tools import PromptGenerator

generator = PromptGenerator()

# 一键生成 prompt
prompt = generator.generate(
    task_description="代码审查",
    examples=[...]
)

# 自动优化
optimized_prompt = generator.optimize(prompt)

架构设计

openjiuwen/
├── core/                    # 核心引擎
│   ├── application/         # Agent 应用
│   ├── workflow/           # 工作流编排
│   ├── single_agent/       # 单 Agent 实现
│   ├── multi_agent/        # 多 Agent 协调
│   ├── runner/             # 执行运行时
│   ├── memory/             # 内存管理
│   ├── retrieval/          # 检索组件
│   ├── session/            # 会话管理
│   └── foundation/         # 基础工具
├── agent_evolving/         # Agent 进化能力
├── deepagents/             # 深度 Agent 实现
├── dev_tools/              # 开发工具
└── extensions/             # 扩展模块

快速开始

安装

pip install openjiuwen

基础示例

import asyncio
from openjiuwen import ReActAgent, LLMConfig

async def main():
    # 配置 LLM
    llm_config = LLMConfig(
        model="gpt-4",
        api_key="your-api-key",
        temperature=0.7
    )
    
    # 创建 Agent
    agent = ReActAgent(
        name="assistant",
        llm_config=llm_config
    )
    
    # 执行任务
    result = await agent.run("帮我分析这个项目的架构")
    print(result)

asyncio.run(main())

高级用法

自定义组件

from openjiuwen.core.foundation import Component

class CustomComponent(Component):
    async def execute(self, input_data):
        # 自定义处理逻辑
        result = await self.process(input_data)
        return result
    
    async def process(self, data):
        # 实现具体处理
        return processed_data

内存管理

from openjiuwen.core.memory import MemoryManager

memory = MemoryManager(
    storage_type="milvus",
    collection_name="agent_memory"
)

# 存储记忆
await memory.store(
    session_id="session_1",
    content="用户偏好：喜欢简洁的回答"
)

# 检索记忆
memories = await memory.retrieve(
    query="用户偏好",
    top_k=5
)

多 Agent 协作

from openjiuwen.core.multi_agent import MultiAgentSystem

system = MultiAgentSystem()

# 添加多个 Agent
system.add_agent("researcher", researcher_agent)
system.add_agent("writer", writer_agent)
system.add_agent("reviewer", reviewer_agent)

# 定义协作流程
system.define_workflow([
    ("researcher", "research"),
    ("writer", "write"),
    ("reviewer", "review")
])

result = await system.execute(task)

设计模式

1. 组件化架构

所有功能通过组件实现，支持灵活组合和扩展。

2. 异步优先

全面支持 async/await，提供高性能并发执行。

3. 图执行模式

基于 DAG 的工作流执行，支持复杂依赖关系。

4. 状态管理

基于检查点的状态持久化，支持中断恢复。

性能优化

并发控制

runner = AsyncGraphRunner(
    max_concurrency=20,  # 最大并发数
    timeout=300,         # 超时时间
    retry_policy={
        "max_retries": 3,
        "backoff": "exponential"
    }
)

流式处理

async for chunk in agent.run_stream(task):
    # 实时处理每个 chunk
    process_chunk(chunk)

最佳实践

1. 使用类型提示: 充分利用 Python 类型系统
2. 环境变量管理: 敏感信息通过环境变量配置
3. 合理设置并发: 根据系统资源调整并发数
4. 监控和日志: 启用详细日志便于调试
5. 测试驱动: 使用 pytest 编写全面测试

常见问题

Q: 如何选择 ReActAgent 和 WorkflowAgent?

A: ReActAgent 适合需要推理和决策的复杂任务，WorkflowAgent 适合流程固定的自动化任务。

Q: 如何处理长时间运行的任务?

A: 使用检查点机制，定期保存状态，支持中断后恢复。

Q: 如何集成自定义 LLM?

A: 实现 LLMInterface 接口，注册到 Agent 配置中。

相关资源

• 官方文档
• GitHub 仓库
• 示例代码

总结

agent-core 提供了构建 AI Agent 的完整基础设施，通过组件化设计、异步执行和丰富的内置功能，让开发者能够快速构建生产级的智能应用。