104 lines
2.7 KiB
Python
104 lines
2.7 KiB
Python
"""
|
|
简单 Agent 示例。
|
|
|
|
演示如何使用 PyOpenClaw 创建一个基本的 Agent。
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
from pyopenclaw.config.settings import AgentConfig
|
|
from pyopenclaw.core.agent import Agent
|
|
from pyopenclaw.core.llm_client import create_llm_client
|
|
from pyopenclaw.core.tool_registry import ToolRegistry
|
|
from pyopenclaw.tools.file_tools import ReadTool, WriteTool
|
|
|
|
|
|
async def main():
|
|
"""运行简单 Agent 示例"""
|
|
# 加载环境变量
|
|
load_dotenv()
|
|
|
|
# 检查 API Key
|
|
api_key = os.getenv("OPENAI_API_KEY")
|
|
if not api_key:
|
|
print("错误: 请设置 OPENAI_API_KEY 环境变量")
|
|
print("示例: export OPENAI_API_KEY='your-api-key'")
|
|
return
|
|
|
|
# 1. 创建 Agent 配置
|
|
config = AgentConfig(
|
|
name="小白",
|
|
model="openai/gpt-4o",
|
|
system_prompt="你是一个有用的 AI 助手，名叫小白。你是一只忠诚的电子宠物狗。",
|
|
tools=["read", "write"],
|
|
max_iterations=10,
|
|
)
|
|
|
|
# 2. 创建 LLM 客户端
|
|
llm_client = create_llm_client(
|
|
provider="openai",
|
|
model="gpt-4o",
|
|
api_key=api_key,
|
|
)
|
|
|
|
# 3. 创建工具注册表并注册工具
|
|
tool_registry = ToolRegistry()
|
|
tool_registry.register(ReadTool())
|
|
tool_registry.register(WriteTool())
|
|
|
|
# 4. 创建 Agent
|
|
agent = Agent(
|
|
config=config,
|
|
llm_client=llm_client,
|
|
tool_registry=tool_registry,
|
|
)
|
|
|
|
print("=" * 50)
|
|
print("PyOpenClaw 简单示例")
|
|
print("=" * 50)
|
|
print(f"Agent 名称: {config.name}")
|
|
print(f"模型: {config.model}")
|
|
print(f"可用工具: {config.tools}")
|
|
print("=" * 50)
|
|
|
|
# 5. 运行 Agent
|
|
user_input = "你好，请介绍一下你自己"
|
|
print(f"\n用户: {user_input}\n")
|
|
print("Agent 回复:")
|
|
print("-" * 50)
|
|
|
|
async for event in agent.run(user_input):
|
|
event_type = event.get("type")
|
|
|
|
if event_type == "content":
|
|
# 内容输出
|
|
print(event["content"])
|
|
if event.get("done"):
|
|
break
|
|
|
|
elif event_type == "tool_call":
|
|
# 工具调用
|
|
print(f"\n[调用工具: {event['name']}]")
|
|
print(f"参数: {event['arguments']}")
|
|
|
|
elif event_type == "tool_result":
|
|
# 工具结果
|
|
result = event["result"]
|
|
if len(result) > 100:
|
|
result = result[:100] + "..."
|
|
print(f"结果: {result}\n")
|
|
|
|
elif event_type == "error":
|
|
# 错误
|
|
print(f"\n错误: {event['content']}")
|
|
break
|
|
|
|
print("-" * 50)
|
|
print("\n运行完成！")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|