AgentOps 帮助开发者构建、评估和监控 AI 智能体。从原型到生产。
AgentOps 应用在 MIT 许可证下开源。您可以在我们的 app 目录 中探索代码。
| 📊 回放分析与调试 | 智能体执行的逐步流程图 |
| 💸 LLM 成本管理 | 追踪与 LLM 基础模型供应商的费用 |
| 🤝 框架集成 | 与 CrewAI、AG2 (AutoGen)、Agno、LangGraph 等原生集成 |
| ⚒️ 自托管 | 想在您自己的云上运行 AgentOps?我们支持 |
pip install agentops
初始化 AgentOps 客户端,即可自动获取所有 LLM 调用的分析数据。
import agentops
# 程序开始处 (例如 main.py, __init__.py)
agentops.init( < 在此处插入您的 API 密钥 >)
...
# 程序结束处
agentops.end_session('Success')
您可以在 AgentOps 控制面板 上查看所有会话。
想在您的机器上运行完整的 AgentOps 应用(控制面板 + API 后端)?请按照 app/README.md 中的设置指南操作:
以最少的代码为您的智能体、工具和函数添加强大的可观测性:一次一行。
请参阅我们的 文档
# 创建会话跨度(所有其他跨度的根)
from agentops.sdk.decorators import session
@session
def my_workflow():
# 您的会话代码写在这里
return result
# 创建智能体跨度以追踪智能体操作
from agentops.sdk.decorators import agent
@agent
class MyAgent:
def __init__(self, name):
self.name = name
# 智能体方法写在这里
# 创建操作/任务跨度以追踪特定操作
from agentops.sdk.decorators import operation, task
@operation # 或 @task
def process_data(data):
# 处理数据
return result
# 创建工作流跨度以追踪多操作工作流
from agentops.sdk.decorators import workflow
@workflow
def my_workflow(data):
# 工作流实现
return result
# 嵌套装饰器以实现正确的跨度层级
from agentops.sdk.decorators import session, agent, operation
@agent
class MyAgent:
@operation
def nested_operation(self, message):
return f"Processed: {message}"
@operation
def main_operation(self):
result = self.nested_operation("test message")
return result
@session
def my_session():
agent = MyAgent()
return agent.main_operation()
所有装饰器均支持:
- 输入/输出记录
- 异常处理
- 异步/等待函数
- 生成器函数
- 自定义属性和名称
使用工具、交接和防护机制构建多智能体系统。AgentOps 原生集成了 OpenAI Agents SDK 的 Python 和 TypeScript 版本。
pip install openai-agents
npm install agentops @openai/agents
仅用两行代码即可构建具有可观测性的 Crew 智能体。只需在环境中设置 AGENTOPS_API_KEY,您的 Crew 就能在 AgentOps 控制面板上获得自动监控。
pip install 'crewai[agentops]'
仅用两行代码,即可为 AG2(原 AutoGen)智能体添加完整的可观测性和监控。在环境中设置 AGENTOPS_API_KEY 并调用 agentops.init()。
通过完整的可观测性追踪和分析 CAMEL 智能体。在环境中设置 AGENTOPS_API_KEY 并初始化 AgentOps 即可开始。
pip install "camel-ai[all]==0.2.11"
pip install agentops
import os
import agentops
from camel.agents import ChatAgent
from camel.messages import BaseMessage
from camel.models import ModelFactory
from camel.types import ModelPlatformType, ModelType
# 初始化 AgentOps
agentops.init(os.getenv("AGENTOPS_API_KEY"), tags=["CAMEL Example"])
# 在 AgentOps 初始化后导入工具包以进行追踪
from camel.toolkits import SearchToolkit
# 使用搜索工具设置智能体
sys_msg = BaseMessage.make_assistant_message(
role_name='Tools calling operator',
content='You are a helpful assistant'
)
# 配置工具和模型
tools = [*SearchToolkit().get_tools()]
model = ModelFactory.create(
model_platform=ModelPlatformType.OPENAI,
model_type=ModelType.GPT_4O_MINI,
)
# 创建并运行智能体
camel_agent = ChatAgent(
system_message=sys_msg,
model=model,
tools=tools,
)
response = camel_agent.step("What is AgentOps?")
print(response)
agentops.end_session("Success")
查看我们的 [Camel 集成指南](https://docs.agentops.ai/v1/integrations/camel) 以获取更多示例,包括多智能体场景。
AgentOps 可与使用 Langchain 构建的应用程序无缝协作。要使用该处理器,请将 Langchain 作为可选依赖项安装:
pip install agentops[langchain]
要使用处理器,请导入并设置
import os
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from agentops.integration.callbacks.langchain import LangchainCallbackHandler
AGENTOPS_API_KEY = os.environ['AGENTOPS_API_KEY']
handler = LangchainCallbackHandler(api_key=AGENTOPS_API_KEY, tags=['Langchain Example'])
llm = ChatOpenAI(openai_api_key=OPENAI_API_KEY,
callbacks=[handler],
model='gpt-3.5-turbo')
agent = initialize_agent(tools,
llm,
agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
callbacks=[handler], # 必须传入回调处理器以记录您的智能体
handle_parsing_errors=True)
查看 [Langchain 示例 Notebook](./examples/langchain/langchain_examples.ipynb) 以获取更多详细信息,包括异步处理器。
为 Cohere (>=5.4.0) 提供一流支持。这是一个持续发展的集成,如果您需要任何新增功能,请在 Discord 上联系我们!
pip install cohere
```python python
import cohere
import agentops
# 程序代码开始处 (例如 main.py, __init__.py)
agentops.init(<在此处插入您的 API 密钥>)
co = cohere.Client()
chat = co.chat(
message="Is it pronounced ceaux-hear or co-hehray?"
)
print(chat)
agentops.end_session('Success')
```python python
import cohere
import agentops
# 程序代码开始处 (例如 main.py, __init__.py)
agentops.init(<在此处插入您的 API 密钥>)
co = cohere.Client()
stream = co.chat_stream(
message="Write me a haiku about the synergies between Cohere and AgentOps"
)
for event in stream:
if event.event_type == "text-generation":
print(event.text, end='')
agentops.end_session('Success')
追踪使用 Anthropic Python SDK (>=0.32.0) 构建的智能体。
pip install anthropic
```python python
import anthropic
import agentops
# 程序代码开始处 (例如 main.py, __init__.py)
agentops.init(<在此处插入您的 API 密钥>)
client = anthropic.Anthropic(
# 这是默认值,可以省略
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
message = client.messages.create(
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Tell me a cool fact about AgentOps",
}
],
model="claude-3-opus-20240229",
)
print(message.content)
agentops.end_session('Success')
```
流式传输
```python python
import anthropic
import agentops
# 程序代码开始处 (例如 main.py, __init__.py)
agentops.init(<在此处插入您的 API 密钥>)
client = anthropic.Anthropic(
# 这是默认值,可以省略
api_key=os