Compare commits

...

11 Commits

Author SHA1 Message Date
Devin AI
95d91d2561 fix: resolve lint and type-checker issues in AgentOps integration
- Fix event handler signatures to match expected (source, event) pattern
- Change 'arguments' to 'tool_args' for tool usage events
- Update TaskEvaluationEvent to use correct attributes (evaluation_type, task)
- Fix corrupted tests/__init__.py file
- Add explicit re-export in src third_party __init__.py

Co-Authored-By: João <joao@crewai.com>
2025-08-18 17:58:59 +00:00
Devin AI
5afe3921d2 feat: restore AgentOps documentation and integration
- Add AgentOps optional dependency to pyproject.toml
- Implement AgentOpsListener with event handling for crew kickoff, tool usage, and task evaluation
- Add comprehensive AgentOps documentation in English, Portuguese, and Korean
- Update docs.json navigation to include AgentOps in observability sections
- Add agentops.log to .gitignore
- Include comprehensive tests for AgentOps integration with mocking
- Ensure graceful handling when AgentOps package is not installed

Addresses issue #3348 by restoring AgentOps integration removed in PR #3334

Co-Authored-By: João <joao@crewai.com>
2025-08-18 17:47:12 +00:00
Greyson LaLonde
1d3d7ebf5e fix: convert XMLSearchTool config values to strings for configparser compatibility (#3344) 2025-08-18 13:23:58 -04:00
Gabe Milani
2c2196f415 fix: flaky test with PytestUnraisableExceptionWarning (#3346) 2025-08-18 14:07:51 -03:00
Gabe Milani
c9f30b175c chore: ignore deprecation warning from chromadb (#3328)
* chore: ignore deprecation warning from chromadb

* adding TODO: in the comment
2025-08-18 13:24:11 -03:00
Greyson LaLonde
a17b93a7f8 Mock telemetry in pytest tests (#3340)
* Add telemetry mocking for pytest tests

- Mock telemetry by default for all tests except telemetry-specific tests
- Add @pytest.mark.telemetry marker for real telemetry tests
- Reduce test overhead and improve isolation

* Fix telemetry test isolation

- Properly isolate telemetry tests from mocking environment
- Preserve API keys and other necessary environment variables
- Ensure telemetry tests can run with real telemetry instances
2025-08-18 11:55:30 -04:00
namho kim
0d3e462791 fix: Revised Korean translation and sentence structure improvement (#3337)
Some checks failed
Notify Downstream / notify-downstream (push) Has been cancelled
2025-08-18 10:46:13 -04:00
Greyson LaLonde
947c9552f0 chore: remove AgentOps integration (#3334)
Some checks failed
Notify Downstream / notify-downstream (push) Has been cancelled
Mark stale issues and pull requests / stale (push) Has been cancelled
2025-08-17 23:07:41 -04:00
Lorenze Jay
04a03d332f Lorenze/emphemeral tracing (#3323)
Some checks failed
Notify Downstream / notify-downstream (push) Has been cancelled
Mark stale issues and pull requests / stale (push) Has been cancelled
* for ephemeral traces

* default false

* simpler and consolidated

* keep raising exception but catch it and continue if its for trace batches

* cleanup

* more cleanup

* not using logger

* refactor: rename TEMP_TRACING_RESOURCE to EPHEMERAL_TRACING_RESOURCE for clarity and consistency in PlusAPI; update related method calls accordingly

* default true

* drop print
2025-08-15 13:37:16 -07:00
Vidit Ostwal
992e093610 Update Docs: Added Mem0 integration with Short Term and Entity Memory (#3293)
Some checks failed
Notify Downstream / notify-downstream (push) Has been cancelled
Mark stale issues and pull requests / stale (push) Has been cancelled
* Added Mem0 integration with Short Term and Entity Memory

* Flaky test case of telemetry
2025-08-14 22:50:24 -04:00
Lucas Gomide
07f8e73958 feat: include exchanged agent messages into ExternalMemory metadata (#3290)
Some checks failed
Notify Downstream / notify-downstream (push) Has been cancelled
2025-08-14 09:41:09 -04:00
41 changed files with 1767 additions and 879 deletions

4
.gitignore vendored
View File

@@ -21,10 +21,12 @@ crew_tasks_output.json
.mypy_cache
.ruff_cache
.venv
agentops.log
test_flow.html
crewairules.mdc
plan.md
conceptual_plan.md
build_image
chromadb-*.lock
# AgentOps
agentops.log

View File

@@ -711,7 +711,7 @@
"icon": "globe"
},
{
"anchor": "법정",
"anchor": "포럼",
"href": "https://community.crewai.com",
"icon": "discourse"
},
@@ -721,7 +721,7 @@
"icon": "robot"
},
{
"anchor": "출시",
"anchor": "릴리스",
"href": "https://github.com/crewAIInc/crewAI/releases",
"icon": "tag"
}
@@ -736,22 +736,22 @@
"pages": ["ko/introduction", "ko/installation", "ko/quickstart"]
},
{
"group": "안내서",
"group": "가이드",
"pages": [
{
"group": "전략",
"pages": ["ko/guides/concepts/evaluating-use-cases"]
},
{
"group": "Agents",
"group": "에이전트 (Agents)",
"pages": ["ko/guides/agents/crafting-effective-agents"]
},
{
"group": "Crews",
"group": "크루 (Crews)",
"pages": ["ko/guides/crews/first-crew"]
},
{
"group": "Flows",
"group": "플로우 (Flows)",
"pages": [
"ko/guides/flows/first-flow",
"ko/guides/flows/mastering-flow-state"
@@ -799,7 +799,7 @@
]
},
{
"group": "도구",
"group": "도구 (Tools)",
"pages": [
"ko/tools/overview",
{
@@ -889,7 +889,7 @@
]
},
{
"group": "클라우드 & 저장",
"group": "클라우드 & 스토리지",
"pages": [
"ko/tools/cloud-storage/overview",
"ko/tools/cloud-storage/s3readertool",
@@ -911,7 +911,7 @@
]
},
{
"group": "오브저버빌리티",
"group": "Observability",
"pages": [
"ko/observability/overview",
"ko/observability/agentops",
@@ -930,7 +930,7 @@
]
},
{
"group": "익히다",
"group": "학습",
"pages": [
"ko/learn/overview",
"ko/learn/llm-selection-guide",
@@ -954,13 +954,13 @@
]
},
{
"group": "원격측정",
"group": "Telemetry",
"pages": ["ko/telemetry"]
}
]
},
{
"tab": "기업",
"tab": "엔터프라이즈",
"groups": [
{
"group": "시작 안내",
@@ -1000,7 +1000,7 @@
]
},
{
"group": "사용 안내서",
"group": "How-To Guides",
"pages": [
"ko/enterprise/guides/build-crew",
"ko/enterprise/guides/deploy-crew",

View File

@@ -177,14 +177,7 @@ class MyCustomCrew:
# Your crew implementation...
```
This is exactly how CrewAI's built-in `agentops_listener` is registered. In the CrewAI codebase, you'll find:
```python
# src/crewai/utilities/events/third_party/__init__.py
from .agentops_listener import agentops_listener
```
This ensures the `agentops_listener` is loaded when the `crewai.utilities.events` package is imported.
This is how third-party event listeners are registered in the CrewAI codebase.
## Available Event Types
@@ -280,77 +273,6 @@ The structure of the event object depends on the event type, but all events inhe
Additional fields vary by event type. For example, `CrewKickoffCompletedEvent` includes `crew_name` and `output` fields.
## Real-World Example: Integration with AgentOps
CrewAI includes an example of a third-party integration with [AgentOps](https://github.com/AgentOps-AI/agentops), a monitoring and observability platform for AI agents. Here's how it's implemented:
```python
from typing import Optional
from crewai.utilities.events import (
CrewKickoffCompletedEvent,
ToolUsageErrorEvent,
ToolUsageStartedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.crew_events import CrewKickoffStartedEvent
from crewai.utilities.events.task_events import TaskEvaluationEvent
try:
import agentops
AGENTOPS_INSTALLED = True
except ImportError:
AGENTOPS_INSTALLED = False
class AgentOpsListener(BaseEventListener):
tool_event: Optional["agentops.ToolEvent"] = None
session: Optional["agentops.Session"] = None
def __init__(self):
super().__init__()
def setup_listeners(self, crewai_event_bus):
if not AGENTOPS_INSTALLED:
return
@crewai_event_bus.on(CrewKickoffStartedEvent)
def on_crew_kickoff_started(source, event: CrewKickoffStartedEvent):
self.session = agentops.init()
for agent in source.agents:
if self.session:
self.session.create_agent(
name=agent.role,
agent_id=str(agent.id),
)
@crewai_event_bus.on(CrewKickoffCompletedEvent)
def on_crew_kickoff_completed(source, event: CrewKickoffCompletedEvent):
if self.session:
self.session.end_session(
end_state="Success",
end_state_reason="Finished Execution",
)
@crewai_event_bus.on(ToolUsageStartedEvent)
def on_tool_usage_started(source, event: ToolUsageStartedEvent):
self.tool_event = agentops.ToolEvent(name=event.tool_name)
if self.session:
self.session.record(self.tool_event)
@crewai_event_bus.on(ToolUsageErrorEvent)
def on_tool_usage_error(source, event: ToolUsageErrorEvent):
agentops.ErrorEvent(exception=event.error, trigger_event=self.tool_event)
```
This listener initializes an AgentOps session when a Crew starts, registers agents with AgentOps, tracks tool usage, and ends the session when the Crew completes.
The AgentOps listener is registered in CrewAI's event system through the import in `src/crewai/utilities/events/third_party/__init__.py`:
```python
from .agentops_listener import agentops_listener
```
This ensures the `agentops_listener` is loaded when the `crewai.utilities.events` package is imported.
## Advanced Usage: Scoped Handlers

View File

@@ -539,16 +539,71 @@ crew = Crew(
)
```
### Mem0 Provider
Short-Term Memory and Entity Memory both supports a tight integration with both Mem0 OSS and Mem0 Client as a provider. Here is how you can use Mem0 as a provider.
```python
from crewai.memory.short_term.short_term_memory import ShortTermMemory
from crewai.memory.entity_entity_memory import EntityMemory
mem0_oss_embedder_config = {
"provider": "mem0",
"config": {
"user_id": "john",
"local_mem0_config": {
"vector_store": {"provider": "qdrant","config": {"host": "localhost", "port": 6333}},
"llm": {"provider": "openai","config": {"api_key": "your-api-key", "model": "gpt-4"}},
"embedder": {"provider": "openai","config": {"api_key": "your-api-key", "model": "text-embedding-3-small"}}
},
"infer": True # Optional defaults to True
},
}
mem0_client_embedder_config = {
"provider": "mem0",
"config": {
"user_id": "john",
"org_id": "my_org_id", # Optional
"project_id": "my_project_id", # Optional
"api_key": "custom-api-key" # Optional - overrides env var
"run_id": "my_run_id", # Optional - for short-term memory
"includes": "include1", # Optional
"excludes": "exclude1", # Optional
"infer": True # Optional defaults to True
"custom_categories": new_categories # Optional - custom categories for user memory
},
}
short_term_memory_mem0_oss = ShortTermMemory(embedder_config=mem0_oss_embedder_config) # Short Term Memory with Mem0 OSS
short_term_memory_mem0_client = ShortTermMemory(embedder_config=mem0_client_embedder_config) # Short Term Memory with Mem0 Client
entity_memory_mem0_oss = EntityMemory(embedder_config=mem0_oss_embedder_config) # Entity Memory with Mem0 OSS
entity_memory_mem0_client = EntityMemory(embedder_config=mem0_client_embedder_config) # Short Term Memory with Mem0 Client
crew = Crew(
memory=True,
short_term_memory=short_term_memory_mem0_oss, # or short_term_memory_mem0_client
entity_memory=entity_memory_mem0_oss # or entity_memory_mem0_client
)
```
### Choosing the Right Embedding Provider
| Provider | Best For | Pros | Cons |
|:---------|:----------|:------|:------|
| **OpenAI** | General use, reliability | High quality, well-tested | Cost, requires API key |
| **Ollama** | Privacy, cost savings | Free, local, private | Requires local setup |
| **Google AI** | Google ecosystem | Good performance | Requires Google account |
| **Azure OpenAI** | Enterprise, compliance | Enterprise features | Complex setup |
| **Cohere** | Multilingual content | Great language support | Specialized use case |
| **VoyageAI** | Retrieval tasks | Optimized for search | Newer provider |
When selecting an embedding provider, consider factors like performance, privacy, cost, and integration needs.
Below is a comparison to help you decide:
| Provider | Best For | Pros | Cons |
| -------------- | ------------------------------ | --------------------------------- | ------------------------- |
| **OpenAI** | General use, high reliability | High quality, widely tested | Paid service, API key required |
| **Ollama** | Privacy-focused, cost savings | Free, runs locally, fully private | Requires local installation/setup |
| **Google AI** | Integration in Google ecosystem| Strong performance, good support | Google account required |
| **Azure OpenAI** | Enterprise & compliance needs| Enterprise-grade features, security | More complex setup process |
| **Cohere** | Multilingual content handling | Excellent language support | More niche use cases |
| **VoyageAI** | Information retrieval & search | Optimized for retrieval tasks | Relatively new provider |
| **Mem0** | Per-user personalization | Search-optimized embeddings | Paid service, API key required |
### Environment Variable Configuration

View File

@@ -1,126 +1,184 @@
---
title: AgentOps Integration
description: Understanding and logging your agent performance with AgentOps.
icon: paperclip
title: "AgentOps Integration"
description: "Monitor and analyze your CrewAI agents with AgentOps observability platform"
---
# Introduction
# AgentOps Integration
Observability is a key aspect of developing and deploying conversational AI agents. It allows developers to understand how their agents are performing,
how their agents are interacting with users, and how their agents use external tools and APIs.
AgentOps is a product independent of CrewAI that provides a comprehensive observability solution for agents.
AgentOps is a powerful observability platform designed specifically for AI agents. It provides comprehensive monitoring, analytics, and debugging capabilities for your CrewAI crews.
## AgentOps
## Features
[AgentOps](https://agentops.ai/?=crew) provides session replays, metrics, and monitoring for agents.
- **Real-time Monitoring**: Track agent performance and behavior in real-time
- **Session Replay**: Review complete agent sessions with detailed execution traces
- **Performance Analytics**: Analyze crew efficiency, tool usage, and task completion rates
- **Error Tracking**: Identify and debug issues in agent workflows
- **Cost Tracking**: Monitor LLM usage and associated costs
- **Team Collaboration**: Share insights and collaborate on agent optimization
At a high level, AgentOps gives you the ability to monitor cost, token usage, latency, agent failures, session-wide statistics, and more.
For more info, check out the [AgentOps Repo](https://github.com/AgentOps-AI/agentops).
## Installation
### Overview
Install AgentOps alongside CrewAI:
AgentOps provides monitoring for agents in development and production.
It provides a dashboard for tracking agent performance, session replays, and custom reporting.
```bash
pip install crewai[agentops]
```
Additionally, AgentOps provides session drilldowns for viewing Crew agent interactions, LLM calls, and tool usage in real-time.
This feature is useful for debugging and understanding how agents interact with users as well as other agents.
Or install AgentOps separately:
![Overview of a select series of agent session runs](/images/agentops-overview.png)
![Overview of session drilldowns for examining agent runs](/images/agentops-session.png)
![Viewing a step-by-step agent replay execution graph](/images/agentops-replay.png)
```bash
pip install agentops
```
### Features
## Setup
- **LLM Cost Management and Tracking**: Track spend with foundation model providers.
- **Replay Analytics**: Watch step-by-step agent execution graphs.
- **Recursive Thought Detection**: Identify when agents fall into infinite loops.
- **Custom Reporting**: Create custom analytics on agent performance.
- **Analytics Dashboard**: Monitor high-level statistics about agents in development and production.
- **Public Model Testing**: Test your agents against benchmarks and leaderboards.
- **Custom Tests**: Run your agents against domain-specific tests.
- **Time Travel Debugging**: Restart your sessions from checkpoints.
- **Compliance and Security**: Create audit logs and detect potential threats such as profanity and PII leaks.
- **Prompt Injection Detection**: Identify potential code injection and secret leaks.
1. **Get your API Key**: Sign up at [AgentOps](https://agentops.ai) and get your API key
### Using AgentOps
2. **Configure your environment**: Set your AgentOps API key as an environment variable:
<Steps>
<Step title="Create an API Key">
Create a user API key here: [Create API Key](https://app.agentops.ai/account)
</Step>
<Step title="Configure Your Environment">
Add your API key to your environment variables:
```bash
AGENTOPS_API_KEY=<YOUR_AGENTOPS_API_KEY>
```
</Step>
<Step title="Install AgentOps">
Install AgentOps with:
```bash
pip install 'crewai[agentops]'
```
or
```bash
pip install agentops
```
</Step>
<Step title="Initialize AgentOps">
Before using `Crew` in your script, include these lines:
```bash
export AGENTOPS_API_KEY="your-api-key-here"
```
```python
import agentops
agentops.init()
```
3. **Initialize AgentOps**: Add this to your CrewAI script:
This will initiate an AgentOps session as well as automatically track Crew agents. For further info on how to outfit more complex agentic systems,
check out the [AgentOps documentation](https://docs.agentops.ai) or join the [Discord](https://discord.gg/j4f3KbeH).
</Step>
</Steps>
```python
import agentops
from crewai import Agent, Task, Crew
### Crew + AgentOps Examples
# Initialize AgentOps
agentops.init()
<CardGroup cols={3}>
<Card
title="Job Posting"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/job-posting"
icon="briefcase"
iconType="solid"
>
Example of a Crew agent that generates job posts.
</Card>
<Card
title="Markdown Validator"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/markdown_validator"
icon="markdown"
iconType="solid"
>
Example of a Crew agent that validates Markdown files.
</Card>
<Card
title="Instagram Post"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/instagram_post"
icon="square-instagram"
iconType="brands"
>
Example of a Crew agent that generates Instagram posts.
</Card>
</CardGroup>
# Your CrewAI code here
agent = Agent(
role="Data Analyst",
goal="Analyze data and provide insights",
backstory="You are an expert data analyst...",
)
### Further Information
task = Task(
description="Analyze the sales data and provide insights",
agent=agent,
)
To get started, create an [AgentOps account](https://agentops.ai/?=crew).
crew = Crew(
agents=[agent],
tasks=[task],
)
For feature requests or bug reports, please reach out to the AgentOps team on the [AgentOps Repo](https://github.com/AgentOps-AI/agentops).
# Run your crew
result = crew.kickoff()
#### Extra links
# End the AgentOps session
agentops.end_session("Success")
```
<a href="https://twitter.com/agentopsai/">🐦 Twitter</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://discord.gg/JHPt4C7r">📢 Discord</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://app.agentops.ai/?=crew">🖇️ AgentOps Dashboard</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://docs.agentops.ai/introduction">📙 Documentation</a>
## Automatic Integration
CrewAI automatically integrates with AgentOps when the library is installed. The integration captures:
- **Crew Kickoff Events**: Start and completion of crew executions
- **Tool Usage**: All tool calls and their results
- **Task Evaluations**: Task performance metrics and feedback
- **Error Events**: Any errors that occur during execution
## Configuration Options
You can customize the AgentOps integration:
```python
import agentops
# Configure AgentOps with custom settings
agentops.init(
api_key="your-api-key",
tags=["production", "data-analysis"],
auto_start_session=True,
instrument_llm_calls=True,
)
```
## Viewing Your Data
1. **Dashboard**: Visit the AgentOps dashboard to view your agent sessions
2. **Session Details**: Click on any session to see detailed execution traces
3. **Analytics**: Use the analytics tab to identify performance trends
4. **Errors**: Monitor the errors tab for debugging information
## Best Practices
- **Tag Your Sessions**: Use meaningful tags to organize your agent runs
- **Monitor Costs**: Keep track of LLM usage and associated costs
- **Review Errors**: Regularly check for and address any errors
- **Optimize Performance**: Use analytics to identify bottlenecks and optimization opportunities
## Troubleshooting
### AgentOps Not Recording Data
1. Verify your API key is set correctly
2. Check that AgentOps is properly initialized
3. Ensure you're calling `agentops.end_session()` at the end of your script
### Missing Events
If some events aren't being captured:
1. Make sure you have the latest version of both CrewAI and AgentOps
2. Check that the AgentOps listener is properly registered
3. Review the logs for any error messages
## Example: Complete Integration
```python
import os
import agentops
from crewai import Agent, Task, Crew, Process
# Initialize AgentOps
agentops.init(
api_key=os.getenv("AGENTOPS_API_KEY"),
tags=["example", "tutorial"],
)
# Define your agents
researcher = Agent(
role="Research Specialist",
goal="Conduct thorough research on given topics",
backstory="You are an expert researcher with access to various tools...",
)
writer = Agent(
role="Content Writer",
goal="Create engaging content based on research",
backstory="You are a skilled writer who can transform research into compelling content...",
)
# Define your tasks
research_task = Task(
description="Research the latest trends in AI and machine learning",
agent=researcher,
)
writing_task = Task(
description="Write a blog post about AI trends based on the research",
agent=writer,
)
# Create and run your crew
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
)
try:
result = crew.kickoff()
print(result)
agentops.end_session("Success")
except Exception as e:
print(f"Error: {e}")
agentops.end_session("Fail")
```
This integration provides comprehensive observability for your CrewAI agents, helping you monitor, debug, and optimize your AI workflows.

View File

@@ -21,9 +21,6 @@ Observability is crucial for understanding how your CrewAI agents perform, ident
### Monitoring & Tracing Platforms
<CardGroup cols={2}>
<Card title="AgentOps" icon="paperclip" href="/en/observability/agentops">
Session replays, metrics, and monitoring for agent development and production.
</Card>
<Card title="LangDB" icon="database" href="/en/observability/langdb">
End-to-end tracing for CrewAI workflows with automatic agent interaction capture.

Binary file not shown.

Before

Width:  |  Height:  |  Size: 288 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 419 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 263 KiB

View File

@@ -177,14 +177,7 @@ class MyCustomCrew:
# Your crew implementation...
```
이것이 바로 CrewAI의 내장 `agentops_listener`가 등록되는 방식과 동일합니다. CrewAI 코드베이스에서는 다음과 같이 되어 있습니다:
```python
# src/crewai/utilities/events/third_party/__init__.py
from .agentops_listener import agentops_listener
```
이렇게 하면 `crewai.utilities.events` 패키지가 임포트될 때 `agentops_listener`가 자동으로 로드됩니다.
이것이 CrewAI 코드베이스에서 서드파티 이벤트 리스너가 등록되는 방식입니다.
## 사용 가능한 이벤트 유형
@@ -280,77 +273,6 @@ CrewAI는 여러분이 청취할 수 있는 다양한 이벤트를 제공합니
추가 필드는 이벤트 타입에 따라 다릅니다. 예를 들어, `CrewKickoffCompletedEvent`에는 `crew_name`과 `output` 필드가 포함됩니다.
## 실제 예시: AgentOps와의 통합
CrewAI는 AI 에이전트를 위한 모니터링 및 관찰 플랫폼인 [AgentOps](https://github.com/AgentOps-AI/agentops)와의 서드파티 통합 예시를 포함하고 있습니다. 구현 방식은 다음과 같습니다:
```python
from typing import Optional
from crewai.utilities.events import (
CrewKickoffCompletedEvent,
ToolUsageErrorEvent,
ToolUsageStartedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.crew_events import CrewKickoffStartedEvent
from crewai.utilities.events.task_events import TaskEvaluationEvent
try:
import agentops
AGENTOPS_INSTALLED = True
except ImportError:
AGENTOPS_INSTALLED = False
class AgentOpsListener(BaseEventListener):
tool_event: Optional["agentops.ToolEvent"] = None
session: Optional["agentops.Session"] = None
def __init__(self):
super().__init__()
def setup_listeners(self, crewai_event_bus):
if not AGENTOPS_INSTALLED:
return
@crewai_event_bus.on(CrewKickoffStartedEvent)
def on_crew_kickoff_started(source, event: CrewKickoffStartedEvent):
self.session = agentops.init()
for agent in source.agents:
if self.session:
self.session.create_agent(
name=agent.role,
agent_id=str(agent.id),
)
@crewai_event_bus.on(CrewKickoffCompletedEvent)
def on_crew_kickoff_completed(source, event: CrewKickoffCompletedEvent):
if self.session:
self.session.end_session(
end_state="Success",
end_state_reason="Finished Execution",
)
@crewai_event_bus.on(ToolUsageStartedEvent)
def on_tool_usage_started(source, event: ToolUsageStartedEvent):
self.tool_event = agentops.ToolEvent(name=event.tool_name)
if self.session:
self.session.record(self.tool_event)
@crewai_event_bus.on(ToolUsageErrorEvent)
def on_tool_usage_error(source, event: ToolUsageErrorEvent):
agentops.ErrorEvent(exception=event.error, trigger_event=self.tool_event)
```
이 listener는 crew가 시작될 때 AgentOps 세션을 초기화하고, agent를 AgentOps에 등록하며, 도구 사용을 추적하고, crew가 완료되면 세션을 종료합니다.
AgentOps listener는 `src/crewai/utilities/events/third_party/__init__.py` 파일의 import를 통해 CrewAI 이벤트 시스템에 등록됩니다:
```python
from .agentops_listener import agentops_listener
```
이렇게 하면 `crewai.utilities.events` 패키지가 import될 때 `agentops_listener`가 로드되는 것이 보장됩니다.
## 고급 사용법: Scoped Handlers

View File

@@ -1,65 +1,65 @@
---
title: 소개
description: 함께 협력하여 복잡한 작업을 해결하는 AI 에이전트 팀 구축
description: 함께 협력하여 복잡한 작업을 해결하는 AI agent 팀 구축
icon: handshake
---
# CrewAI란 무엇인가?
**CrewAI는 완전히 독립적으로, LangChain이나 기타 agent 프레임워크에 의존하지 않고 처음부터 스크래치로 개발된 가볍고 매우 빠른 Python 프레임워크입니다.**
**CrewAI는 LangChain이나 기타 agent 프레임워크에 의존하지 않고, 완전히 독립적으로 처음부터 스크래치로 개발된 가볍고 매우 빠른 Python 프레임워크입니다.**
CrewAI는 고수준의 간편함과 정밀한 저수준 제어를 모두 제공하여, 어떤 시나리오에도 맞춤화된 자율 AI agent를 만드는 데 이상적입니다:
- **[CrewAI Crews](/ko/guides/crews/first-crew)**: 자율성과 협업 지능을 극대화하여, 각 agent가 특정 역할, 도구, 목표를 가진 AI 팀을 만들 수 있습니다.
- **[CrewAI Flows](/ko/guides/flows/first-flow)**: 세밀한 이벤트 기반 제어와 단일 LLM 호출을 통한 정확한 작업 오케스트레이션을 가능하게 하며 Crews 네이티브로 지원합니다.
- **[CrewAI Flows](/ko/guides/flows/first-flow)**: 이벤트 기반의 세밀한 제어와 단일 LLM 호출을 통한 정확한 작업 orchestration을 지원하며, Crews 네이티브로 통합됩니다.
10만 명이 넘는 개발자가 커뮤니티 과정을 통해 인증을 받았으며, CrewAI는 기업용 AI 자동화의 표준으로 빠르게 자리잡고 있습니다.
## 크루 작동 방식
## Crew의 작동 방식
<Note>
회사가 비즈니스 목표를 달성하기 위해 여러 부서(영업, 엔지니어링, 마케팅 등)가 리더십 아래에서 함께 일하는 것처럼, CrewAI는 복잡한 작업을 달성하기 위해 전문화된 역할의 AI 에이전트들이 협력하는 조직을 만들 수 있도록 도와줍니다.
회사가 비즈니스 목표를 달성하기 위해 여러 부서(영업, 엔지니어링, 마케팅 등)가 리더십 아래에서 함께 일하는 것처럼, CrewAI는 복잡한 작업을 달성하기 위해 전문화된 역할의 AI agent들이 협력하는 조직을 만들 수 있도록 도와줍니다.
</Note>
<Frame caption="CrewAI 프레임워크 개요">
<Frame caption="CrewAI Framework Overview">
<img src="/images/crews.png" alt="CrewAI Framework Overview" />
</Frame>
| 구성 요소 | 설명 | 주요 특징 |
|:--------------|:---------------------:|:----------|
| **크루** | 최상위 조직 | • AI 에이전트 팀 관리<br/>• 워크플로우 감독<br/>• 협업 보장<br/>• 결과 전달 |
| **AI 에이전트** | 전문 팀원 | • 특정 역할 보유(연구원, 작가 등)<br/>• 지정된 도구 사용<br/>• 작업 위임 가능<br/>• 자율적 의사결정 가능 |
| **프로세스** | 워크플로우 관리 시스템 | • 협업 패턴 정의<br/>• 작업 할당 제어<br/>• 상호작용 관리<br/>• 효율적 실행 보장 |
| **작업** | 개별 할당 | • 명확한 목표 보유<br/>• 특정 도구 사용<br/>• 더 큰 프로세스에 기여<br/>• 실행 가능한 결과 도출 |
| 구성 요소 | 설명 | 주요 특징 |
|:----------|:----:|:----------|
| **Crew** | 최상위 조직 | • AI agent 팀 관리<br/>• workflow 감독<br/>• 협업 보장<br/>• 결과 전달 |
| **AI agents** | 전문 팀원 | • 특정 역할 보유(Researcher, Writer 등)<br/>• 지정된 도구 사용<br/>• 작업 위임 가능<br/>• 자율적 의사결정 가능 |
| **Process** | workflow 관리 시스템 | • 협업 패턴 정의<br/>• 작업 할당 제어<br/>• 상호작용 관리<br/>• 효율적 실행 보장 |
| **Task** | 개별 할당 | • 명확한 목표 보유<br/>• 특정 도구 사용<br/>• 더 큰 프로세스에 기여<br/>• 실행 가능한 결과 도출 |
### 어떻게 모두 함께 작동하는가
### 전체 구조의 동작 방식
1. **Crew**가 전체 운영을 조직합니다
2. **AI Agents**가 자신들의 전문 작업을 수행합니다
2. **AI agents**가 자신들의 전문 작업을 수행합니다
3. **Process**가 원활한 협업을 보장합니다
4. **Tasks**가 완료되어 목표를 달성합니다
## 주요 기능
<CardGroup cols={2}>
<Card title="역할 기반 에이전트" icon="users">
연구원, 분석가, 작가 등 다양한 역할, 전문성, 목표를 가진 전문 에이전트를 생성할 수 있습니다
<Card title="역할 기반 agent" icon="users">
Researcher, Analyst, Writer 등 다양한 역할 전문성, 목표를 가진 agent를 생성할 수 있습니다
</Card>
<Card title="유연한 도구" icon="screwdriver-wrench">
에이전트에게 외부 서비스 및 데이터 소스와 상호작용할 수 있는 맞춤형 도구와 API를 제공합니다
agent에게 외부 서비스 및 데이터 소스와 상호작용할 수 있는 맞춤형 도구와 API를 제공합니다
</Card>
<Card title="지능형 협업" icon="people-arrows">
에이전트가 함께 작업하며, 인사이트를 공유하고 작업을 조율하여 복잡한 목표를 달성합니다
agent들이 함께 작업하며, 인사이트를 공유하고 작업을 조율하여 복잡한 목표를 달성합니다
</Card>
<Card title="작업 관리" icon="list-check">
순차적 또는 병렬 워크플로우를 정의할 수 있으며, 에이전트가 작업 의존성을 자동으로 처리합니다
순차적 또는 병렬 workflow를 정의할 수 있으며, agent가 작업 의존성을 자동으로 처리합니다
</Card>
</CardGroup>
## 플로우의 작동 원리
## Flow의 작동 원리
<Note>
crew 자율 협업에 탁월한 반면, 플로우는 구조화된 자동화를 제공하여 워크플로우 실행에 대한 세밀한 제어를 제공합니다. 플로우는 조건부 로직, 반복문, 동적 상태 관리를 정확하게 처리하면서 작업이 신뢰성 있게, 안전하게, 효율적으로 실행되도록 보장합니다. 플로우crew와 원활하게 통합되어 높은 자율성과 엄격한 제어의 균형을 이룰 수 있게 해줍니다.
Crew 자율 협업에 탁월하다면, Flow는 구조화된 자동화를 제공하여 workflow 실행에 대한 세밀한 제어를 제공합니다. Flow는 조건부 로직, 반복문, 동적 상태 관리를 정확하게 처리하면서 작업이 신뢰성 있게, 안전하게, 효율적으로 실행되도록 보장합니다. FlowCrew와 원활하게 통합되어 높은 자율성과 엄격한 제어의 균형을 이룰 수 있게 해줍니다.
</Note>
<Frame caption="CrewAI Framework Overview">
@@ -68,41 +68,41 @@ CrewAI는 고수준의 간편함과 정밀한 저수준 제어를 모두 제공
| 구성 요소 | 설명 | 주요 기능 |
|:----------|:-----------:|:------------|
| **Flow** | 구조화된 워크플로우 오케스트레이션 | • 실행 경로 관리<br/>• 상태 전환 처리<br/>• 작업 순서 제어<br/>• 신뢰성 있는 실행 보장 |
| **Events** | 워크플로우 액션 트리거 | • 특정 프로세스 시작<br/>• 동적 응답 가능<br/>• 조건부 분기 지원<br/>• 실시간 적응 허용 |
| **States** | 워크플로우 실행 컨텍스트 | • 실행 데이터 유지<br/>• 데이터 영속성 지원<br/>• 재개 가능성 보장<br/>• 실행 무결성 확보 |
| **Crew Support** | 워크플로우 자동화 강화 | • 필요할 때 agency 삽입<br/>• 구조화된 워크플로우 보완<br/>• 자동화와 인텔리전스의 균형<br/>• 적응적 의사결정 지원 |
| **Flow** | 구조화된 workflow orchestration | • 실행 경로 관리<br/>• 상태 전환 처리<br/>• 작업 순서 제어<br/>• 신뢰성 있는 실행 보장 |
| **Events** | workflow 액션 트리거 | • 특정 프로세스 시작<br/>• 동적 응답 가능<br/>• 조건부 분기 지원<br/>• 실시간 적응 허용 |
| **States** | workflow 실행 컨텍스트 | • 실행 데이터 유지<br/>• 데이터 영속성 지원<br/>• 재개 가능성 보장<br/>• 실행 무결성 확보 |
| **Crew Support** | workflow 자동화 강화 | • 필요할 때 agency 삽입<br/>• 구조화된 workflow 보완<br/>• 자동화와 인텔리전스의 균형<br/>• 적응적 의사결정 지원 |
### 주요 기능
<CardGroup cols={2}>
<Card title="이벤트 기반 오케스트레이션" icon="bolt">
이벤트에 동적으로 반응하여 정밀한 실행 경로 정의
<Card title="이벤트 기반 orchestration" icon="bolt">
이벤트에 동적으로 반응하여 정밀한 실행 경로 정의합니다
</Card>
<Card title="세밀한 제어" icon="sliders">
워크플로우 상태와 조건부 실행을 안전하고 효율적으로 관리
workflow 상태와 조건부 실행을 안전하고 효율적으로 관리합니다
</Card>
<Card title="네이티브 Crew 통합" icon="puzzle-piece">
Crews와 손쉽게 결합하여 자율성과 지능 강화
Crews와 손쉽게 결합하여 자율성과 지능 강화합니다
</Card>
<Card title="결정론적 실행" icon="route">
명시적 제어 흐름과 오류 처리로 예측 가능한 결과 보장
명시적 제어 흐름과 오류 처리로 예측 가능한 결과 보장합니다
</Card>
</CardGroup>
## 크루(Crews)와 플로우(Flows)를 언제 사용할까
## CrewFlow를 언제 사용할까
<Note>
[크루](/ko/guides/crews/first-crew)와 [플로우](/ko/guides/flows/first-flow)를 언제 사용할지 이해하는 것은 CrewAI의 잠재력을 애플리케이션에서 극대화하는 데 핵심적입니다.
[Crew](/ko/guides/crews/first-crew)와 [Flow](/ko/guides/flows/first-flow)를 언제 사용할지 이해하는 것은 CrewAI의 잠재력을 애플리케이션에서 극대화하는 데 핵심적입니다.
</Note>
| 사용 사례 | 권장 접근 방식 | 이유 |
|:---------|:---------------------|:-----|
| **개방형 연구** | [크루](/ko/guides/crews/first-crew) | 과제가 창의적 사고, 탐색, 적응이 필요할 때 |
| **콘텐츠 생성** | [크루](/ko/guides/crews/first-crew) | 기사, 보고서, 마케팅 자료 등 협업형 생성 |
| **의사결정 워크플로우** | [플로우](/ko/guides/flows/first-flow) | 예측 가능하고 감사 가능한 의사결정 경로 및 정밀 제어가 필요할 때 |
| **API 오케스트레이션** | [플로우](/ko/guides/flows/first-flow) | 특정 순서로 여러 외부 서비스에 신뢰성 있게 통합할 때 |
| **하이브리드 애플리케이션** | 혼합 접근 방식 | [플로우](/ko/guides/flows/first-flow)로 전체 프로세스를 오케스트레이션하고, [크루](/ko/guides/crews/first-crew)로 복잡한 하위 작업을 처리 |
| **개방형 연구** | [Crew](/ko/guides/crews/first-crew) | 창의적 사고, 탐색, 적응이 필요한 작업에 적합 |
| **콘텐츠 생성** | [Crew](/ko/guides/crews/first-crew) | 기사, 보고서, 마케팅 자료 등 협업형 생성에 적합 |
| **의사결정 workflow** | [Flow](/ko/guides/flows/first-flow) | 예측 가능하고 감사 가능한 의사결정 경로 및 정밀 제어가 필요할 때 |
| **API orchestration** | [Flow](/ko/guides/flows/first-flow) | 특정 순서로 여러 외부 서비스에 신뢰성 있게 통합할 때 |
| **하이브리드 애플리케이션** | 혼합 접근 방식 | [Flow](/ko/guides/flows/first-flow)로 전체 프로세스를 orchestration하고, [Crew](/ko/guides/crews/first-crew)로 복잡한 하위 작업을 처리 |
### 의사결정 프레임워크
@@ -112,8 +112,8 @@ CrewAI는 고수준의 간편함과 정밀한 저수준 제어를 모두 제공
## CrewAI를 선택해야 하는 이유?
- 🧠 **자율적 운영**: 에이전트가 자신의 역할과 사용 가능한 도구를 바탕으로 지능적인 결정을 내립니다
- 📝 **자연스러운 상호작용**: 에이전트가 인간 팀원처럼 소통하고 협업합니다
- 🧠 **자율적 운영**: agent가 자신의 역할과 사용 가능한 도구를 바탕으로 지능적인 결정을 내립니다
- 📝 **자연스러운 상호작용**: agent가 인간 팀원처럼 소통하고 협업합니다
- 🛠️ **확장 가능한 설계**: 새로운 도구, 역할, 기능을 쉽게 추가할 수 있습니다
- 🚀 **프로덕션 준비 완료**: 실제 환경에서의 신뢰성과 확장성을 고려하여 구축되었습니다
- 🔒 **보안 중심**: 엔터프라이즈 보안 요구 사항을 고려하여 설계되었습니다
@@ -134,7 +134,7 @@ CrewAI는 고수준의 간편함과 정밀한 저수준 제어를 모두 제공
icon="diagram-project"
href="/ko/guides/flows/first-flow"
>
실행을 정밀하게 제어할 수 있는 구조화된, 이벤트 기반 워크플로우를 만드는 방법을 배워보세요.
실행을 정밀하게 제어할 수 있는 구조화된, 이벤트 기반 workflow를 만드는 방법을 배워보세요.
</Card>
</CardGroup>
@@ -151,7 +151,7 @@ CrewAI는 고수준의 간편함과 정밀한 저수준 제어를 모두 제공
icon="bolt"
href="ko/quickstart"
>
빠른 시작 가이드를 따라 첫 번째 CrewAI 에이전트를 만들고 직접 경험해 보세요.
빠른 시작 가이드를 따라 첫 번째 CrewAI agent를 만들고 직접 경험해 보세요.
</Card>
<Card
title="커뮤니티 가입하기"

View File

@@ -1,124 +1,131 @@
---
title: AgentOps 통합
description: AgentOps를 사용하여 에이전트 성능을 이해하고 로깅하기
icon: paperclip
title: "AgentOps 통합"
description: "AgentOps 관찰 가능성 플랫폼으로 CrewAI 에이전트를 모니터링하고 분석하세요"
---
# 소개
# AgentOps 통합
Observability는 대화형 AI 에이전트를 개발하고 배포하는 데 있어 핵심적인 요소입니다. 이는 개발자가 에이전트의 성능을 이해하고, 에이전트가 사용자와 어떻게 상호작용하는지, 그리고 에이전트가 외부 도구와 API를 어떻게 사용하는지를 파악할 수 있게 해줍니다.
AgentOps는 CrewAI와 독립적인 제품으로, 에이전트를 위한 종합적인 observability 솔루션을 제공합니다.
AgentOps는 AI 에이전트를 위해 특별히 설계된 강력한 관찰 가능성 플랫폼입니다. CrewAI 크루를 위한 포괄적인 모니터링, 분석 및 디버깅 기능을 제공합니다.
## AgentOps
## 기능
[AgentOps](https://agentops.ai/?=crew)은 에이전트에 대한 세션 리플레이, 메트릭, 모니터링을 제공합니다.
- **실시간 모니터링**: 에이전트 성능과 동작을 실시간으로 추적
- **세션 재생**: 상세한 실행 추적과 함께 완전한 에이전트 세션 검토
- **성능 분석**: 크루 효율성, 도구 사용량 및 작업 완료율 분석
- **오류 추적**: 에이전트 워크플로우의 문제 식별 및 디버그
- **비용 추적**: LLM 사용량 및 관련 비용 모니터링
- **팀 협업**: 인사이트 공유 및 에이전트 최적화 협업
AgentOps는 높은 수준에서 비용, 토큰 사용량, 대기 시간, 에이전트 실패, 세션 전체 통계 등 다양한 항목을 모니터링할 수 있는 기능을 제공합니다.
더 자세한 내용은 [AgentOps Repo](https://github.com/AgentOps-AI/agentops)를 확인하세요.
## 설치
### 개요
CrewAI와 함께 AgentOps 설치:
AgentOps는 개발 및 프로덕션 환경에서 에이전트에 대한 모니터링을 제공합니다.
에이전트 성능, 세션 리플레이, 맞춤형 리포팅을 추적할 수 있는 대시보드를 제공합니다.
```bash
pip install crewai[agentops]
```
한, AgentOps는 Crew 에이전트 상호작용, LLM 호출, 툴 사용을 실시간으로 볼 수 있는 세션 드릴다운 기능을 제공합니다.
이 기능은 에이전트가 사용자 및 다른 에이전트와 어떻게 상호작용하는지 디버깅하고 이해하는 데 유용합니다.
AgentOps를 별도로 설치:
![선택된 에이전트 세션 실행 시리즈의 개요](/images/agentops-overview.png)
![에이전트 실행을 조사하기 위한 세션 드릴다운 개요](/images/agentops-session.png)
![단계별 에이전트 리플레이 실행 그래프 보기](/images/agentops-replay.png)
```bash
pip install agentops
```
### 특징
## 설정
- **LLM 비용 관리 및 추적**: 기반 모델 공급자와의 지출을 추적합니다.
- **재생 분석**: 단계별 에이전트 실행 그래프를 시청할 수 있습니다.
- **재귀적 사고 감지**: 에이전트가 무한 루프에 빠졌는지 식별합니다.
- **맞춤형 보고서**: 에이전트 성능에 대한 맞춤형 분석을 생성합니다.
- **분석 대시보드**: 개발 및 운영 중인 에이전트에 대한 상위 수준 통계를 모니터링합니다.
- **공개 모델 테스트**: 벤치마크 및 리더보드를 통해 에이전트를 테스트할 수 있습니다.
- **맞춤형 테스트**: 도메인별 테스트로 에이전트를 실행합니다.
- **타임 트래블 디버깅**: 체크포인트에서 세션을 재시작합니다.
- **컴플라이언스 및 보안**: 감사 로그를 생성하고 욕설 및 PII 유출과 같은 잠재적 위협을 감지합니다.
- **프롬프트 인젝션 감지**: 잠재적 코드 인젝션 및 시크릿 유출을 식별합니다.
1. **API 키 받기**: [AgentOps](https://agentops.ai)에 가입하고 API 키를 받으세요
### AgentOps 사용하기
2. **환경 구성**: AgentOps API 키를 환경 변수로 설정:
<Steps>
<Step title="API 키 생성">
사용자 API 키를 여기서 생성하세요: [API 키 생성](https://app.agentops.ai/account)
</Step>
<Step title="환경 설정">
API 키를 환경 변수에 추가하세요:
```bash
AGENTOPS_API_KEY=<YOUR_AGENTOPS_API_KEY>
```
</Step>
<Step title="AgentOps 설치">
다음 명령어로 AgentOps를 설치하세요:
```bash
pip install 'crewai[agentops]'
```
또는
```bash
pip install agentops
```
</Step>
<Step title="AgentOps 초기화">
스크립트에서 `Crew`를 사용하기 전에 다음 코드를 포함하세요:
```bash
export AGENTOPS_API_KEY="여기에-api-키-입력"
```
```python
import agentops
agentops.init()
```
3. **AgentOps 초기화**: CrewAI 스크립트에 다음을 추가:
이렇게 하면 AgentOps 세션이 시작되고 Crew 에이전트가 자동으로 추적됩니다. 더 복잡한 agentic 시스템을 구성하는 방법에 대한 자세한 정보는 [AgentOps 문서](https://docs.agentops.ai) 또는 [Discord](https://discord.gg/j4f3KbeH)를 참조하세요.
</Step>
</Steps>
```python
import agentops
from crewai import Agent, Task, Crew
### Crew + AgentOps 예시
# AgentOps 초기화
agentops.init()
<CardGroup cols={3}>
<Card
title="Job Posting"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/job-posting"
icon="briefcase"
iconType="solid"
>
채용 공고를 생성하는 Crew agent의 예시입니다.
</Card>
<Card
title="Markdown Validator"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/markdown_validator"
icon="markdown"
iconType="solid"
>
Markdown 파일을 검증하는 Crew agent의 예시입니다.
</Card>
<Card
title="Instagram Post"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/instagram_post"
icon="square-instagram"
iconType="brands"
>
Instagram 게시물을 생성하는 Crew agent의 예시입니다.
</Card>
</CardGroup>
# 여기에 CrewAI 코드
agent = Agent(
role="데이터 분석가",
goal="데이터를 분석하고 인사이트 제공",
backstory="당신은 전문 데이터 분석가입니다...",
)
### 추가 정보
task = Task(
description="판매 데이터를 분석하고 인사이트를 제공하세요",
agent=agent,
)
시작하려면 [AgentOps 계정](https://agentops.ai/?=crew)을 생성하세요.
crew = Crew(
agents=[agent],
tasks=[task],
)
기능 요청이나 버그 보고가 필요하시면 [AgentOps Repo](https://github.com/AgentOps-AI/agentops)에서 AgentOps 팀에 문의해 주세요.
# 크루 실행
result = crew.kickoff()
#### 추가 링크
# AgentOps 세션 종료
agentops.end_session("Success")
```
<a href="https://twitter.com/agentopsai/">🐦 트위터</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://discord.gg/JHPt4C7r">📢 디스코드</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://app.agentops.ai/?=crew">🖇️ AgentOps 대시보드</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://docs.agentops.ai/introduction">📙 문서화</a>
## 자동 통합
CrewAI는 라이브러리가 설치되면 AgentOps와 자동으로 통합됩니다. 통합은 다음을 캡처합니다:
- **크루 킥오프 이벤트**: 크루 실행의 시작과 완료
- **도구 사용**: 모든 도구 호출과 결과
- **작업 평가**: 작업 성능 메트릭과 피드백
- **오류 이벤트**: 실행 중 발생하는 모든 오류
## 구성 옵션
AgentOps 통합을 사용자 정의할 수 있습니다:
```python
import agentops
# 사용자 정의 설정으로 AgentOps 구성
agentops.init(
api_key="당신의-api-키",
tags=["프로덕션", "데이터-분석"],
auto_start_session=True,
instrument_llm_calls=True,
)
```
## 데이터 보기
1. **대시보드**: AgentOps 대시보드를 방문하여 에이전트 세션 보기
2. **세션 세부사항**: 세션을 클릭하여 상세한 실행 추적 보기
3. **분석**: 분석 탭을 사용하여 성능 트렌드 식별
4. **오류**: 디버깅 정보를 위해 오류 탭 모니터링
## 모범 사례
- **세션 태그 지정**: 의미 있는 태그를 사용하여 에이전트 실행 정리
- **비용 모니터링**: LLM 사용량과 관련 비용 추적
- **오류 검토**: 정기적으로 오류 확인 및 해결
- **성능 최적화**: 분석을 사용하여 병목 현상과 최적화 기회 식별
## 문제 해결
### AgentOps가 데이터를 기록하지 않음
1. API 키가 올바르게 설정되었는지 확인
2. AgentOps가 제대로 초기화되었는지 확인
3. 스크립트 끝에서 `agentops.end_session()`을 호출하는지 확인
### 누락된 이벤트
일부 이벤트가 캡처되지 않는 경우:
1. CrewAI와 AgentOps의 최신 버전이 있는지 확인
2. AgentOps 리스너가 제대로 등록되었는지 확인
3. 오류 메시지에 대한 로그 검토
이 통합은 CrewAI 에이전트에 대한 포괄적인 관찰 가능성을 제공하여 AI 워크플로우를 모니터링, 디버그 및 최적화하는 데 도움이 됩니다.

View File

@@ -21,9 +21,6 @@ icon: "face-smile"
### 모니터링 & 트레이싱 플랫폼
<CardGroup cols={2}>
<Card title="AgentOps" icon="paperclip" href="/ko/observability/agentops">
에이전트 개발 및 운영을 위한 세션 리플레이, 메트릭, 모니터링 제공.
</Card>
<Card title="LangDB" icon="database" href="/ko/observability/langdb">
자동 에이전트 상호작용 캡처를 포함한 CrewAI 워크플로의 엔드-투-엔드 트레이싱.

View File

@@ -177,14 +177,7 @@ class MyCustomCrew:
# Sua implementação do crew...
```
É exatamente assim que o `agentops_listener` integrado do CrewAI é registrado. No código-fonte do CrewAI, você encontrará:
```python
# src/crewai/utilities/events/third_party/__init__.py
from .agentops_listener import agentops_listener
```
Isso garante que o `agentops_listener` seja carregado quando o pacote `crewai.utilities.events` for importado.
É assim que listeners de eventos de terceiros são registrados no código do CrewAI.
## Tipos de Eventos Disponíveis
@@ -269,77 +262,6 @@ A estrutura do objeto de evento depende do tipo do evento, mas todos herdam de `
Campos adicionais variam pelo tipo de evento. Por exemplo, `CrewKickoffCompletedEvent` inclui os campos `crew_name` e `output`.
## Exemplo Real: Integração com AgentOps
O CrewAI inclui um exemplo de integração com [AgentOps](https://github.com/AgentOps-AI/agentops), uma plataforma de monitoramento e observabilidade para agentes de IA. Veja como é implementado:
```python
from typing import Optional
from crewai.utilities.events import (
CrewKickoffCompletedEvent,
ToolUsageErrorEvent,
ToolUsageStartedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.crew_events import CrewKickoffStartedEvent
from crewai.utilities.events.task_events import TaskEvaluationEvent
try:
import agentops
AGENTOPS_INSTALLED = True
except ImportError:
AGENTOPS_INSTALLED = False
class AgentOpsListener(BaseEventListener):
tool_event: Optional["agentops.ToolEvent"] = None
session: Optional["agentops.Session"] = None
def __init__(self):
super().__init__()
def setup_listeners(self, crewai_event_bus):
if not AGENTOPS_INSTALLED:
return
@crewai_event_bus.on(CrewKickoffStartedEvent)
def on_crew_kickoff_started(source, event: CrewKickoffStartedEvent):
self.session = agentops.init()
for agent in source.agents:
if self.session:
self.session.create_agent(
name=agent.role,
agent_id=str(agent.id),
)
@crewai_event_bus.on(CrewKickoffCompletedEvent)
def on_crew_kickoff_completed(source, event: CrewKickoffCompletedEvent):
if self.session:
self.session.end_session(
end_state="Success",
end_state_reason="Finished Execution",
)
@crewai_event_bus.on(ToolUsageStartedEvent)
def on_tool_usage_started(source, event: ToolUsageStartedEvent):
self.tool_event = agentops.ToolEvent(name=event.tool_name)
if self.session:
self.session.record(self.tool_event)
@crewai_event_bus.on(ToolUsageErrorEvent)
def on_tool_usage_error(source, event: ToolUsageErrorEvent):
agentops.ErrorEvent(exception=event.error, trigger_event=self.tool_event)
```
Esse listener inicializa uma sessão do AgentOps quando um Crew inicia, cadastra agentes no AgentOps, rastreia o uso de ferramentas e finaliza a sessão quando o Crew é concluído.
O listener AgentOps é registrado no sistema de eventos do CrewAI via importação em `src/crewai/utilities/events/third_party/__init__.py`:
```python
from .agentops_listener import agentops_listener
```
Isso garante que o `agentops_listener` seja carregado quando o pacote `crewai.utilities.events` for importado.
## Uso Avançado: Handlers Escopados

View File

@@ -1,126 +1,131 @@
---
title: Integração com AgentOps
description: Entendendo e registrando a performance do seu agente com AgentOps.
icon: paperclip
title: "Integração AgentOps"
description: "Monitore e analise seus agentes CrewAI com a plataforma de observabilidade AgentOps"
---
# Introdução
# Integração AgentOps
Observabilidade é um aspecto fundamental no desenvolvimento e implantação de agentes de IA conversacional. Ela permite que desenvolvedores compreendam como seus agentes estão performando,
como eles estão interagindo com os usuários e como utilizam ferramentas externas e APIs.
AgentOps é um produto independente do CrewAI que fornece uma solução completa de observabilidade para agentes.
AgentOps é uma poderosa plataforma de observabilidade projetada especificamente para agentes de IA. Ela fornece capacidades abrangentes de monitoramento, análise e depuração para suas crews CrewAI.
## AgentOps
## Recursos
[AgentOps](https://agentops.ai/?=crew) oferece replay de sessões, métricas e monitoramento para agentes.
- **Monitoramento em Tempo Real**: Acompanhe o desempenho e comportamento dos agentes em tempo real
- **Replay de Sessão**: Revise sessões completas de agentes com rastreamentos detalhados de execução
- **Análise de Desempenho**: Analise eficiência da crew, uso de ferramentas e taxas de conclusão de tarefas
- **Rastreamento de Erros**: Identifique e depure problemas em fluxos de trabalho de agentes
- **Rastreamento de Custos**: Monitore o uso de LLM e custos associados
- **Colaboração em Equipe**: Compartilhe insights e colabore na otimização de agentes
Em um alto nível, o AgentOps oferece a capacidade de monitorar custos, uso de tokens, latência, falhas do agente, estatísticas de sessão e muito mais.
Para mais informações, confira o [Repositório do AgentOps](https://github.com/AgentOps-AI/agentops).
## Instalação
### Visão Geral
Instale o AgentOps junto com o CrewAI:
AgentOps fornece monitoramento para agentes em desenvolvimento e produção.
Disponibiliza um dashboard para acompanhamento de performance dos agentes, replay de sessões e relatórios personalizados.
```bash
pip install crewai[agentops]
```
Além disso, o AgentOps traz análises detalhadas das sessões para visualizar interações do agente Crew, chamadas LLM e uso de ferramentas em tempo real.
Esse recurso é útil para depuração e entendimento de como os agentes interagem com usuários e entre si.
Ou instale o AgentOps separadamente:
![Visão geral de uma série selecionada de execuções de sessões do agente](/images/agentops-overview.png)
![Visão geral das análises detalhadas de sessões para examinar execuções de agentes](/images/agentops-session.png)
![Visualizando um gráfico de execução passo a passo do replay do agente](/images/agentops-replay.png)
```bash
pip install agentops
```
### Funcionalidades
## Configuração
- **Gerenciamento e Rastreamento de Custos de LLM**: Acompanhe gastos com provedores de modelos fundamentais.
- **Análises de Replay**: Assista gráficos de execução do agente, passo a passo.
- **Detecção de Pensamento Recursivo**: Identifique quando agentes entram em loops infinitos.
- **Relatórios Personalizados**: Crie análises customizadas sobre a performance dos agentes.
- **Dashboard Analítico**: Monitore estatísticas gerais de agentes em desenvolvimento e produção.
- **Teste de Modelos Públicos**: Teste seus agentes em benchmarks e rankings.
- **Testes Personalizados**: Execute seus agentes em testes específicos de domínio.
- **Depuração com Viagem no Tempo**: Reinicie suas sessões a partir de checkpoints.
- **Conformidade e Segurança**: Crie registros de auditoria e detecte possíveis ameaças como uso de palavrões e vazamento de dados pessoais.
- **Detecção de Prompt Injection**: Identifique possíveis injeções de código e vazamentos de segredos.
1. **Obtenha sua Chave API**: Cadastre-se no [AgentOps](https://agentops.ai) e obtenha sua chave API
### Utilizando o AgentOps
2. **Configure seu ambiente**: Defina sua chave API do AgentOps como uma variável de ambiente:
<Steps>
<Step title="Crie uma Chave de API">
Crie uma chave de API de usuário aqui: [Create API Key](https://app.agentops.ai/account)
</Step>
<Step title="Configure seu Ambiente">
Adicione sua chave API nas variáveis de ambiente:
```bash
AGENTOPS_API_KEY=<YOUR_AGENTOPS_API_KEY>
```
</Step>
<Step title="Instale o AgentOps">
Instale o AgentOps com:
```bash
pip install 'crewai[agentops]'
```
ou
```bash
pip install agentops
```
</Step>
<Step title="Inicialize o AgentOps">
Antes de utilizar o `Crew` no seu script, inclua estas linhas:
```bash
export AGENTOPS_API_KEY="sua-chave-api-aqui"
```
```python
import agentops
agentops.init()
```
3. **Inicialize o AgentOps**: Adicione isso ao seu script CrewAI:
Isso irá iniciar uma sessão do AgentOps e também rastrear automaticamente os agentes Crew. Para mais detalhes sobre como adaptar sistemas de agentes mais complexos,
confira a [documentação do AgentOps](https://docs.agentops.ai) ou participe do [Discord](https://discord.gg/j4f3KbeH).
</Step>
</Steps>
```python
import agentops
from crewai import Agent, Task, Crew
### Exemplos de Crew + AgentOps
# Inicializar AgentOps
agentops.init()
<CardGroup cols={3}>
<Card
title="Vaga de Emprego"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/job-posting"
icon="briefcase"
iconType="solid"
>
Exemplo de um agente Crew que gera vagas de emprego.
</Card>
<Card
title="Validador de Markdown"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/markdown_validator"
icon="markdown"
iconType="solid"
>
Exemplo de um agente Crew que valida arquivos Markdown.
</Card>
<Card
title="Post no Instagram"
color="#F3A78B"
href="https://github.com/joaomdmoura/crewAI-examples/tree/main/instagram_post"
icon="square-instagram"
iconType="brands"
>
Exemplo de um agente Crew que gera posts para Instagram.
</Card>
</CardGroup>
# Seu código CrewAI aqui
agent = Agent(
role="Analista de Dados",
goal="Analisar dados e fornecer insights",
backstory="Você é um analista de dados especialista...",
)
### Mais Informações
task = Task(
description="Analise os dados de vendas e forneça insights",
agent=agent,
)
Para começar, crie uma [conta AgentOps](https://agentops.ai/?=crew).
crew = Crew(
agents=[agent],
tasks=[task],
)
Para sugestões de funcionalidades ou relatos de bugs, entre em contato com o time do AgentOps pelo [Repositório do AgentOps](https://github.com/AgentOps-AI/agentops).
# Execute sua crew
result = crew.kickoff()
#### Links Extras
# Finalize a sessão AgentOps
agentops.end_session("Success")
```
<a href="https://twitter.com/agentopsai/">🐦 Twitter</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://discord.gg/JHPt4C7r">📢 Discord</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://app.agentops.ai/?=crew">🖇️ Dashboard AgentOps</a>
<span>&nbsp;&nbsp;•&nbsp;&nbsp;</span>
<a href="https://docs.agentops.ai/introduction">📙 Documentação</a>
## Integração Automática
O CrewAI se integra automaticamente com o AgentOps quando a biblioteca está instalada. A integração captura:
- **Eventos de Kickoff da Crew**: Início e conclusão de execuções da crew
- **Uso de Ferramentas**: Todas as chamadas de ferramentas e seus resultados
- **Avaliações de Tarefas**: Métricas de desempenho de tarefas e feedback
- **Eventos de Erro**: Quaisquer erros que ocorram durante a execução
## Opções de Configuração
Você pode personalizar a integração do AgentOps:
```python
import agentops
# Configure AgentOps com configurações personalizadas
agentops.init(
api_key="sua-chave-api",
tags=["producao", "analise-dados"],
auto_start_session=True,
instrument_llm_calls=True,
)
```
## Visualizando Seus Dados
1. **Dashboard**: Visite o dashboard do AgentOps para ver suas sessões de agentes
2. **Detalhes da Sessão**: Clique em qualquer sessão para ver rastreamentos detalhados de execução
3. **Análises**: Use a aba de análises para identificar tendências de desempenho
4. **Erros**: Monitore a aba de erros para informações de depuração
## Melhores Práticas
- **Marque Suas Sessões**: Use tags significativas para organizar suas execuções de agentes
- **Monitore Custos**: Acompanhe o uso de LLM e custos associados
- **Revise Erros**: Verifique e resolva regularmente quaisquer erros
- **Otimize Desempenho**: Use análises para identificar gargalos e oportunidades de otimização
## Solução de Problemas
### AgentOps Não Está Gravando Dados
1. Verifique se sua chave API está definida corretamente
2. Verifique se o AgentOps está inicializado adequadamente
3. Certifique-se de estar chamando `agentops.end_session()` no final do seu script
### Eventos Ausentes
Se alguns eventos não estão sendo capturados:
1. Certifique-se de ter a versão mais recente do CrewAI e AgentOps
2. Verifique se o listener do AgentOps está registrado adequadamente
3. Revise os logs para quaisquer mensagens de erro
Esta integração fornece observabilidade abrangente para seus agentes CrewAI, ajudando você a monitorar, depurar e otimizar seus fluxos de trabalho de IA.

View File

@@ -21,9 +21,6 @@ A observabilidade é fundamental para entender como seus agentes CrewAI estão d
### Plataformas de Monitoramento e Rastreamento
<CardGroup cols={2}>
<Card title="AgentOps" icon="paperclip" href="/pt-BR/observability/agentops">
Replays de sessões, métricas e monitoramento para desenvolvimento e produção de agentes.
</Card>
<Card title="LangDB" icon="database" href="/pt-BR/observability/langdb">
Rastreamento ponta a ponta para fluxos de trabalho CrewAI com captura automática de interações de agentes.

View File

@@ -52,7 +52,6 @@ tools = ["crewai-tools~=0.62.0"]
embeddings = [
"tiktoken~=0.8.0"
]
agentops = ["agentops==0.3.18"]
pdfplumber = [
"pdfplumber>=0.11.4",
]
@@ -69,6 +68,7 @@ docling = [
aisuite = [
"aisuite>=0.1.10",
]
agentops = ["agentops==0.3.18"]
[tool.uv]
dev-dependencies = [
@@ -99,6 +99,11 @@ exclude = ["cli/templates"]
[tool.bandit]
exclude_dirs = ["src/crewai/cli/templates"]
[tool.pytest.ini_options]
markers = [
"telemetry: mark test as a telemetry test (don't mock telemetry)",
]
# PyTorch index configuration, since torch 2.5.0 is not compatible with python 3.13
[[tool.uv.index]]
name = "pytorch-nightly"

View File

@@ -1,5 +1,5 @@
import time
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Dict, List
from crewai.memory.entity.entity_memory_item import EntityMemoryItem
from crewai.memory.long_term.long_term_memory_item import LongTermMemoryItem
@@ -21,6 +21,7 @@ class CrewAgentExecutorMixin:
task: "Task"
iterations: int
max_iter: int
messages: List[Dict[str, str]]
_i18n: I18N
_printer: Printer = Printer()
@@ -62,6 +63,7 @@ class CrewAgentExecutorMixin:
value=output.text,
metadata={
"description": self.task.description,
"messages": self.messages,
},
agent=self.agent.role,
)
@@ -127,7 +129,6 @@ class CrewAgentExecutorMixin:
def _ask_human_input(self, final_answer: str) -> str:
"""Prompt human input with mode-appropriate messaging."""
event_listener.formatter.pause_live_updates()
try:
self._printer.print(
content=f"\033[1m\033[95m ## Final Result:\033[00m \033[92m{final_answer}\033[00m"

View File

@@ -1,9 +1,13 @@
from .utils import TokenManager
class AuthError(Exception):
pass
def get_auth_token() -> str:
"""Get the authentication token."""
access_token = TokenManager().get_token()
if not access_token:
raise Exception("No token found, make sure you are logged in")
raise AuthError("No token found, make sure you are logged in")
return access_token

View File

@@ -18,6 +18,7 @@ class PlusAPI:
CREWS_RESOURCE = "/crewai_plus/api/v1/crews"
AGENTS_RESOURCE = "/crewai_plus/api/v1/agents"
TRACING_RESOURCE = "/crewai_plus/api/v1/tracing"
EPHEMERAL_TRACING_RESOURCE = "/crewai_plus/api/v1/tracing/ephemeral"
def __init__(self, api_key: str) -> None:
self.api_key = api_key
@@ -124,6 +125,11 @@ class PlusAPI:
"POST", f"{self.TRACING_RESOURCE}/batches", json=payload
)
def initialize_ephemeral_trace_batch(self, payload) -> requests.Response:
return self._make_request(
"POST", f"{self.EPHEMERAL_TRACING_RESOURCE}/batches", json=payload
)
def send_trace_events(self, trace_batch_id: str, payload) -> requests.Response:
return self._make_request(
"POST",
@@ -131,9 +137,27 @@ class PlusAPI:
json=payload,
)
def send_ephemeral_trace_events(
self, trace_batch_id: str, payload
) -> requests.Response:
return self._make_request(
"POST",
f"{self.EPHEMERAL_TRACING_RESOURCE}/batches/{trace_batch_id}/events",
json=payload,
)
def finalize_trace_batch(self, trace_batch_id: str, payload) -> requests.Response:
return self._make_request(
"PATCH",
f"{self.TRACING_RESOURCE}/batches/{trace_batch_id}/finalize",
json=payload,
)
def finalize_ephemeral_trace_batch(
self, trace_batch_id: str, payload
) -> requests.Response:
return self._make_request(
"PATCH",
f"{self.EPHEMERAL_TRACING_RESOURCE}/batches/{trace_batch_id}/finalize",
json=payload,
)

View File

@@ -77,7 +77,10 @@ from crewai.utilities.events.listeners.tracing.trace_listener import (
)
from crewai.utilities.events.listeners.tracing.utils import is_tracing_enabled
from crewai.utilities.events.listeners.tracing.utils import (
is_tracing_enabled,
on_first_execution_tracing_confirmation,
)
from crewai.utilities.formatter import (
aggregate_raw_outputs_from_task_outputs,
aggregate_raw_outputs_from_tasks,
@@ -283,8 +286,11 @@ class Crew(FlowTrackable, BaseModel):
self._cache_handler = CacheHandler()
event_listener = EventListener()
if on_first_execution_tracing_confirmation():
self.tracing = True
if is_tracing_enabled() or self.tracing:
trace_listener = TraceCollectionListener(tracing=self.tracing)
trace_listener = TraceCollectionListener()
trace_listener.setup_listeners(crewai_event_bus)
event_listener.verbose = self.verbose
event_listener.formatter.verbose = self.verbose

View File

@@ -38,7 +38,10 @@ from crewai.utilities.events.flow_events import (
from crewai.utilities.events.listeners.tracing.trace_listener import (
TraceCollectionListener,
)
from crewai.utilities.events.listeners.tracing.utils import is_tracing_enabled
from crewai.utilities.events.listeners.tracing.utils import (
is_tracing_enabled,
on_first_execution_tracing_confirmation,
)
from crewai.utilities.printer import Printer
logger = logging.getLogger(__name__)
@@ -476,8 +479,12 @@ class Flow(Generic[T], metaclass=FlowMeta):
# Initialize state with initial values
self._state = self._create_initial_state()
self.tracing = tracing
if is_tracing_enabled() or tracing:
trace_listener = TraceCollectionListener(tracing=tracing)
if (
on_first_execution_tracing_confirmation()
or is_tracing_enabled()
or self.tracing
):
trace_listener = TraceCollectionListener()
trace_listener.setup_listeners(crewai_event_bus)
# Apply any additional kwargs
if kwargs:

View File

@@ -11,6 +11,8 @@ import chromadb.errors
from chromadb.api import ClientAPI
from chromadb.api.types import OneOrMany
from chromadb.config import Settings
from pydantic.warnings import PydanticDeprecatedSince211
import warnings
from crewai.knowledge.storage.base_knowledge_storage import BaseKnowledgeStorage
from crewai.rag.embeddings.configurator import EmbeddingConfigurator
@@ -85,6 +87,15 @@ class KnowledgeStorage(BaseKnowledgeStorage):
raise Exception("Collection not initialized")
def initialize_knowledge_storage(self):
# Suppress deprecation warnings from chromadb, which are not relevant to us
# TODO: Remove this once we upgrade chromadb to at least 1.0.8.
warnings.filterwarnings(
"ignore",
category=PydanticDeprecatedSince211,
message=r".*'model_fields'.*is deprecated.*",
module=r"^chromadb(\.|$)",
)
self.app = create_persistent_client(
path=os.path.join(db_storage_path(), "knowledge"),
settings=Settings(allow_reset=True),

View File

@@ -12,6 +12,8 @@ from crewai.rag.embeddings.configurator import EmbeddingConfigurator
from crewai.utilities.chromadb import create_persistent_client
from crewai.utilities.constants import MAX_FILE_NAME_LENGTH
from crewai.utilities.paths import db_storage_path
import warnings
from pydantic.warnings import PydanticDeprecatedSince211
@contextlib.contextmanager
@@ -62,6 +64,15 @@ class RAGStorage(BaseRAGStorage):
def _initialize_app(self):
from chromadb.config import Settings
# Suppress deprecation warnings from chromadb, which are not relevant to us
# TODO: Remove this once we upgrade chromadb to at least 1.0.8.
warnings.filterwarnings(
"ignore",
category=PydanticDeprecatedSince211,
message=r".*'model_fields'.*is deprecated.*",
module=r"^chromadb(\.|$)",
)
self._set_embedder_config()
self.app = create_persistent_client(

View File

@@ -71,7 +71,6 @@ from .third_party.agentops_listener import agentops_listener
__all__ = [
"EventListener",
"agentops_listener",
"CrewAIEventsBus",
"crewai_event_bus",
"AgentExecutionStartedEvent",
@@ -105,7 +104,6 @@ __all__ = [
"MemoryRetrievalStartedEvent",
"MemoryRetrievalCompletedEvent",
"EventListener",
"agentops_listener",
"CrewKickoffStartedEvent",
"CrewKickoffCompletedEvent",
"CrewKickoffFailedEvent",
@@ -124,4 +122,5 @@ __all__ = [
"ToolSelectionErrorEvent",
"ToolUsageEvent",
"ToolValidateInputErrorEvent",
"agentops_listener",
]

View File

@@ -4,7 +4,7 @@ from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from crewai.utilities.constants import CREWAI_BASE_URL
from crewai.cli.authentication.token import get_auth_token
from crewai.cli.authentication.token import AuthError, get_auth_token
from crewai.cli.version import get_crewai_version
from crewai.cli.plus_api import PlusAPI
@@ -41,14 +41,21 @@ class TraceBatchManager:
"""Single responsibility: Manage batches and event buffering"""
def __init__(self):
self.plus_api = PlusAPI(api_key=get_auth_token())
try:
self.plus_api = PlusAPI(api_key=get_auth_token())
except AuthError:
self.plus_api = PlusAPI(api_key="")
self.trace_batch_id: Optional[str] = None # Backend ID
self.current_batch: Optional[TraceBatch] = None
self.event_buffer: List[TraceEvent] = []
self.execution_start_times: Dict[str, datetime] = {}
def initialize_batch(
self, user_context: Dict[str, str], execution_metadata: Dict[str, Any]
self,
user_context: Dict[str, str],
execution_metadata: Dict[str, Any],
use_ephemeral: bool = False,
) -> TraceBatch:
"""Initialize a new trace batch"""
self.current_batch = TraceBatch(
@@ -57,13 +64,15 @@ class TraceBatchManager:
self.event_buffer.clear()
self.record_start_time("execution")
self._initialize_backend_batch(user_context, execution_metadata)
self._initialize_backend_batch(user_context, execution_metadata, use_ephemeral)
return self.current_batch
def _initialize_backend_batch(
self, user_context: Dict[str, str], execution_metadata: Dict[str, Any]
self,
user_context: Dict[str, str],
execution_metadata: Dict[str, Any],
use_ephemeral: bool = False,
):
"""Send batch initialization to backend"""
@@ -74,6 +83,7 @@ class TraceBatchManager:
payload = {
"trace_id": self.current_batch.batch_id,
"execution_type": execution_metadata.get("execution_type", "crew"),
"user_identifier": execution_metadata.get("user_context", None),
"execution_context": {
"crew_fingerprint": execution_metadata.get("crew_fingerprint"),
"crew_name": execution_metadata.get("crew_name", None),
@@ -91,12 +101,22 @@ class TraceBatchManager:
"execution_started_at": datetime.now(timezone.utc).isoformat(),
},
}
if use_ephemeral:
payload["ephemeral_trace_id"] = self.current_batch.batch_id
response = self.plus_api.initialize_trace_batch(payload)
response = (
self.plus_api.initialize_ephemeral_trace_batch(payload)
if use_ephemeral
else self.plus_api.initialize_trace_batch(payload)
)
if response.status_code == 201 or response.status_code == 200:
response_data = response.json()
self.trace_batch_id = response_data["trace_id"]
self.trace_batch_id = (
response_data["trace_id"]
if not use_ephemeral
else response_data["ephemeral_trace_id"]
)
console = Console()
panel = Panel(
f"✅ Trace batch initialized with session ID: {self.trace_batch_id}",
@@ -116,7 +136,7 @@ class TraceBatchManager:
"""Add event to buffer"""
self.event_buffer.append(trace_event)
def _send_events_to_backend(self):
def _send_events_to_backend(self, ephemeral: bool = True):
"""Send buffered events to backend"""
if not self.plus_api or not self.trace_batch_id or not self.event_buffer:
return
@@ -134,7 +154,11 @@ class TraceBatchManager:
if not self.trace_batch_id:
raise Exception("❌ Trace batch ID not found")
response = self.plus_api.send_trace_events(self.trace_batch_id, payload)
response = (
self.plus_api.send_ephemeral_trace_events(self.trace_batch_id, payload)
if ephemeral
else self.plus_api.send_trace_events(self.trace_batch_id, payload)
)
if response.status_code == 200 or response.status_code == 201:
self.event_buffer.clear()
@@ -146,15 +170,15 @@ class TraceBatchManager:
except Exception as e:
logger.error(f"❌ Error sending events to backend: {str(e)}")
def finalize_batch(self) -> Optional[TraceBatch]:
def finalize_batch(self, ephemeral: bool = True) -> Optional[TraceBatch]:
"""Finalize batch and return it for sending"""
if not self.current_batch:
return None
if self.event_buffer:
self._send_events_to_backend()
self._send_events_to_backend(ephemeral)
self._finalize_backend_batch()
self._finalize_backend_batch(ephemeral)
self.current_batch.events = self.event_buffer.copy()
@@ -168,7 +192,7 @@ class TraceBatchManager:
return finalized_batch
def _finalize_backend_batch(self):
def _finalize_backend_batch(self, ephemeral: bool = True):
"""Send batch finalization to backend"""
if not self.plus_api or not self.trace_batch_id:
return
@@ -182,12 +206,24 @@ class TraceBatchManager:
"final_event_count": total_events,
}
response = self.plus_api.finalize_trace_batch(self.trace_batch_id, payload)
response = (
self.plus_api.finalize_ephemeral_trace_batch(
self.trace_batch_id, payload
)
if ephemeral
else self.plus_api.finalize_trace_batch(self.trace_batch_id, payload)
)
if response.status_code == 200:
access_code = response.json().get("access_code", None)
console = Console()
return_link = (
f"{CREWAI_BASE_URL}/crewai_plus/trace_batches/{self.trace_batch_id}"
if not ephemeral and access_code
else f"{CREWAI_BASE_URL}/crewai_plus/ephemeral_trace_batches/{self.trace_batch_id}?access_code={access_code}"
)
panel = Panel(
f"✅ Trace batch finalized with session ID: {self.trace_batch_id}. View here: {CREWAI_BASE_URL}/crewai_plus/trace_batches/{self.trace_batch_id}",
f"✅ Trace batch finalized with session ID: {self.trace_batch_id}. View here: {return_link} {f', Access Code: {access_code}' if access_code else ''}",
title="Trace Batch Finalization",
border_style="green",
)

View File

@@ -13,7 +13,6 @@ from crewai.utilities.events.agent_events import (
AgentExecutionErrorEvent,
)
from crewai.utilities.events.listeners.tracing.types import TraceEvent
from crewai.utilities.events.listeners.tracing.utils import is_tracing_enabled
from crewai.utilities.events.reasoning_events import (
AgentReasoningStartedEvent,
AgentReasoningCompletedEvent,
@@ -67,7 +66,7 @@ from crewai.utilities.events.memory_events import (
MemorySaveFailedEvent,
)
from crewai.cli.authentication.token import get_auth_token
from crewai.cli.authentication.token import AuthError, get_auth_token
from crewai.cli.version import get_crewai_version
@@ -76,13 +75,12 @@ class TraceCollectionListener(BaseEventListener):
Trace collection listener that orchestrates trace collection
"""
trace_enabled: Optional[bool] = False
complex_events = ["task_started", "llm_call_started", "llm_call_completed"]
_instance = None
_initialized = False
def __new__(cls, batch_manager=None, tracing: Optional[bool] = False):
def __new__(cls, batch_manager=None):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
@@ -90,25 +88,22 @@ class TraceCollectionListener(BaseEventListener):
def __init__(
self,
batch_manager: Optional[TraceBatchManager] = None,
tracing: Optional[bool] = False,
):
if self._initialized:
return
super().__init__()
self.batch_manager = batch_manager or TraceBatchManager()
self.tracing = tracing or False
self.trace_enabled = self._check_trace_enabled()
self._initialized = True
def _check_trace_enabled(self) -> bool:
def _check_authenticated(self) -> bool:
"""Check if tracing should be enabled"""
auth_token = get_auth_token()
if not auth_token:
try:
res = bool(get_auth_token())
return res
except AuthError:
return False
return is_tracing_enabled() or self.tracing
def _get_user_context(self) -> Dict[str, str]:
"""Extract user context for tracing"""
return {
@@ -120,8 +115,6 @@ class TraceCollectionListener(BaseEventListener):
def setup_listeners(self, crewai_event_bus):
"""Setup event listeners - delegates to specific handlers"""
if not self.trace_enabled:
return
self._register_flow_event_handlers(crewai_event_bus)
self._register_context_event_handlers(crewai_event_bus)
@@ -167,13 +160,13 @@ class TraceCollectionListener(BaseEventListener):
@event_bus.on(CrewKickoffStartedEvent)
def on_crew_started(source, event):
if not self.batch_manager.is_batch_initialized():
self._initialize_batch(source, event)
self._initialize_crew_batch(source, event)
self._handle_trace_event("crew_kickoff_started", source, event)
@event_bus.on(CrewKickoffCompletedEvent)
def on_crew_completed(source, event):
self._handle_trace_event("crew_kickoff_completed", source, event)
self.batch_manager.finalize_batch()
self.batch_manager.finalize_batch(ephemeral=True)
@event_bus.on(CrewKickoffFailedEvent)
def on_crew_failed(source, event):
@@ -287,7 +280,7 @@ class TraceCollectionListener(BaseEventListener):
def on_agent_reasoning_failed(source, event):
self._handle_action_event("agent_reasoning_failed", source, event)
def _initialize_batch(self, source: Any, event: Any):
def _initialize_crew_batch(self, source: Any, event: Any):
"""Initialize trace batch"""
user_context = self._get_user_context()
execution_metadata = {
@@ -296,7 +289,7 @@ class TraceCollectionListener(BaseEventListener):
"crewai_version": get_crewai_version(),
}
self.batch_manager.initialize_batch(user_context, execution_metadata)
self._initialize_batch(user_context, execution_metadata)
def _initialize_flow_batch(self, source: Any, event: Any):
"""Initialize trace batch for Flow execution"""
@@ -308,7 +301,20 @@ class TraceCollectionListener(BaseEventListener):
"execution_type": "flow",
}
self.batch_manager.initialize_batch(user_context, execution_metadata)
self._initialize_batch(user_context, execution_metadata)
def _initialize_batch(
self, user_context: Dict[str, str], execution_metadata: Dict[str, Any]
):
"""Initialize trace batch if ephemeral"""
if not self._check_authenticated():
self.batch_manager.initialize_batch(
user_context, execution_metadata, use_ephemeral=True
)
else:
self.batch_manager.initialize_batch(
user_context, execution_metadata, use_ephemeral=False
)
def _handle_trace_event(self, event_type: str, source: Any, event: Any):
"""Generic handler for context end events"""

View File

@@ -1,5 +1,153 @@
import os
import platform
import uuid
import hashlib
import subprocess
import getpass
from pathlib import Path
from datetime import datetime
import re
import json
import click
from crewai.utilities.paths import db_storage_path
def is_tracing_enabled() -> bool:
return os.getenv("CREWAI_TRACING_ENABLED", "false").lower() == "true"
def on_first_execution_tracing_confirmation() -> bool:
if _is_test_environment():
return False
if is_first_execution():
mark_first_execution_done()
return click.confirm(
"This is the first execution of CrewAI. Do you want to enable tracing?",
default=True,
show_default=True,
)
return False
def _is_test_environment() -> bool:
"""Detect if we're running in a test environment."""
return os.environ.get("CREWAI_TESTING", "").lower() == "true"
def _get_machine_id() -> str:
"""Stable, privacy-preserving machine fingerprint (cross-platform)."""
parts = []
try:
mac = ":".join(
["{:02x}".format((uuid.getnode() >> b) & 0xFF) for b in range(0, 12, 2)][
::-1
]
)
parts.append(mac)
except Exception:
pass
sysname = platform.system()
parts.append(sysname)
try:
if sysname == "Darwin":
res = subprocess.run(
["system_profiler", "SPHardwareDataType"],
capture_output=True,
text=True,
timeout=2,
)
m = re.search(r"Hardware UUID:\s*([A-Fa-f0-9\-]+)", res.stdout)
if m:
parts.append(m.group(1))
elif sysname == "Linux":
try:
parts.append(Path("/etc/machine-id").read_text().strip())
except Exception:
parts.append(Path("/sys/class/dmi/id/product_uuid").read_text().strip())
elif sysname == "Windows":
res = subprocess.run(
["wmic", "csproduct", "get", "UUID"],
capture_output=True,
text=True,
timeout=2,
)
lines = [line.strip() for line in res.stdout.splitlines() if line.strip()]
if len(lines) >= 2:
parts.append(lines[1])
except Exception:
pass
return hashlib.sha256("".join(parts).encode()).hexdigest()
def _user_data_file() -> Path:
base = Path(db_storage_path())
base.mkdir(parents=True, exist_ok=True)
return base / ".crewai_user.json"
def _load_user_data() -> dict:
p = _user_data_file()
if p.exists():
try:
return json.loads(p.read_text())
except Exception:
pass
return {}
def _save_user_data(data: dict) -> None:
try:
p = _user_data_file()
p.write_text(json.dumps(data, indent=2))
except Exception:
pass
def get_user_id() -> str:
"""Stable, anonymized user identifier with caching."""
data = _load_user_data()
if "user_id" in data:
return data["user_id"]
try:
username = getpass.getuser()
except Exception:
username = "unknown"
seed = f"{username}|{_get_machine_id()}"
uid = hashlib.sha256(seed.encode()).hexdigest()
data["user_id"] = uid
_save_user_data(data)
return uid
def is_first_execution() -> bool:
"""True if this is the first execution for this user."""
data = _load_user_data()
return not data.get("first_execution_done", False)
def mark_first_execution_done() -> None:
"""Mark that the first execution has been completed."""
data = _load_user_data()
if data.get("first_execution_done", False):
return
data.update(
{
"first_execution_done": True,
"first_execution_at": datetime.now().timestamp(),
"user_id": get_user_id(),
"machine_id": _get_machine_id(),
}
)
_save_user_data(data)

View File

@@ -1 +1 @@
from .agentops_listener import agentops_listener
from .agentops_listener import agentops_listener as agentops_listener

View File

@@ -1,67 +1,137 @@
from typing import Optional
import logging
from crewai.utilities.events import (
from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.crewai_event_bus import CrewAIEventsBus
from crewai.utilities.events.crew_events import (
CrewKickoffCompletedEvent,
CrewKickoffStartedEvent,
)
from crewai.utilities.events.task_events import TaskEvaluationEvent
from crewai.utilities.events.tool_usage_events import (
ToolUsageErrorEvent,
ToolUsageStartedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.crew_events import CrewKickoffStartedEvent
from crewai.utilities.events.task_events import TaskEvaluationEvent
try:
import agentops
AGENTOPS_INSTALLED = True
except ImportError:
AGENTOPS_INSTALLED = False
logger = logging.getLogger(__name__)
class AgentOpsListener(BaseEventListener):
tool_event: Optional["agentops.ToolEvent"] = None
session: Optional["agentops.Session"] = None
def __init__(self):
self.agentops = None
try:
import agentops
self.agentops = agentops
logger.info("AgentOps integration enabled")
except ImportError:
logger.debug("AgentOps not installed, skipping AgentOps integration")
super().__init__()
def setup_listeners(self, crewai_event_bus):
if not AGENTOPS_INSTALLED:
def setup_listeners(self, crewai_event_bus: CrewAIEventsBus):
if self.agentops is None:
return
@crewai_event_bus.on(CrewKickoffStartedEvent)
def on_crew_kickoff_started(source, event: CrewKickoffStartedEvent):
self.session = agentops.init()
for agent in source.agents:
if self.session:
self.session.create_agent(
name=agent.role,
agent_id=str(agent.id),
)
def on_crew_kickoff_started(source, event):
self._handle_crew_kickoff_started(source, event)
@crewai_event_bus.on(CrewKickoffCompletedEvent)
def on_crew_kickoff_completed(source, event: CrewKickoffCompletedEvent):
if self.session:
self.session.end_session(
end_state="Success",
end_state_reason="Finished Execution",
)
def on_crew_kickoff_completed(source, event):
self._handle_crew_kickoff_completed(source, event)
@crewai_event_bus.on(ToolUsageStartedEvent)
def on_tool_usage_started(source, event: ToolUsageStartedEvent):
self.tool_event = agentops.ToolEvent(name=event.tool_name)
if self.session:
self.session.record(self.tool_event)
def on_tool_usage_started(source, event):
self._handle_tool_usage_started(source, event)
@crewai_event_bus.on(ToolUsageErrorEvent)
def on_tool_usage_error(source, event: ToolUsageErrorEvent):
agentops.ErrorEvent(exception=event.error, trigger_event=self.tool_event)
def on_tool_usage_error(source, event):
self._handle_tool_usage_error(source, event)
@crewai_event_bus.on(TaskEvaluationEvent)
def on_task_evaluation(source, event: TaskEvaluationEvent):
if self.session:
self.session.create_agent(
name="Task Evaluator", agent_id=str(source.original_agent.id)
def on_task_evaluation(source, event):
self._handle_task_evaluation(source, event)
def _handle_crew_kickoff_started(self, source, event: CrewKickoffStartedEvent):
if self.agentops is None:
return
try:
self.agentops.start_session(
tags=["crewai", "crew_kickoff"],
config=self.agentops.Configuration(
auto_start_session=False,
instrument_llm_calls=True,
),
)
logger.debug("AgentOps session started for crew kickoff")
except Exception as e:
logger.warning(f"Failed to start AgentOps session: {e}")
def _handle_crew_kickoff_completed(self, source, event: CrewKickoffCompletedEvent):
if self.agentops is None:
return
try:
self.agentops.end_session("Success")
logger.debug("AgentOps session ended for crew kickoff completion")
except Exception as e:
logger.warning(f"Failed to end AgentOps session: {e}")
def _handle_tool_usage_started(self, source, event: ToolUsageStartedEvent):
if self.agentops is None:
return
try:
self.agentops.record(
self.agentops.ActionEvent(
action_type="tool_usage",
params={
"tool_name": event.tool_name,
"tool_args": event.tool_args,
},
)
)
logger.debug(f"AgentOps recorded tool usage: {event.tool_name}")
except Exception as e:
logger.warning(f"Failed to record tool usage in AgentOps: {e}")
def _handle_tool_usage_error(self, source, event: ToolUsageErrorEvent):
if self.agentops is None:
return
try:
self.agentops.record(
self.agentops.ErrorEvent(
message=f"Tool usage error: {event.error}",
error_type="ToolUsageError",
details={
"tool_name": event.tool_name,
"tool_args": event.tool_args,
},
)
)
logger.debug(f"AgentOps recorded tool usage error: {event.tool_name}")
except Exception as e:
logger.warning(f"Failed to record tool usage error in AgentOps: {e}")
def _handle_task_evaluation(self, source, event: TaskEvaluationEvent):
if self.agentops is None:
return
try:
self.agentops.record(
self.agentops.ActionEvent(
action_type="task_evaluation",
params={
"evaluation_type": event.evaluation_type,
"task": str(event.task) if event.task else None,
},
)
)
logger.debug(f"AgentOps recorded task evaluation: {event.evaluation_type}")
except Exception as e:
logger.warning(f"Failed to record task evaluation in AgentOps: {e}")
agentops_listener = AgentOpsListener()

View File

@@ -1896,7 +1896,7 @@ def test_agent_with_knowledge_sources_generate_search_query():
assert "red" in result.raw.lower()
@pytest.mark.vcr(record_mode='none', filter_headers=["authorization"])
@pytest.mark.vcr(record_mode="none", filter_headers=["authorization"])
def test_agent_with_knowledge_with_no_crewai_knowledge():
mock_knowledge = MagicMock(spec=Knowledge)
@@ -1904,8 +1904,11 @@ def test_agent_with_knowledge_with_no_crewai_knowledge():
role="Information Agent",
goal="Provide information based on knowledge sources",
backstory="You have access to specific knowledge sources.",
llm=LLM(model="openrouter/openai/gpt-4o-mini",api_key=os.getenv('OPENROUTER_API_KEY')),
knowledge=mock_knowledge
llm=LLM(
model="openrouter/openai/gpt-4o-mini",
api_key=os.getenv("OPENROUTER_API_KEY"),
),
knowledge=mock_knowledge,
)
# Create a task that requires the agent to use the knowledge
@@ -1920,7 +1923,7 @@ def test_agent_with_knowledge_with_no_crewai_knowledge():
mock_knowledge.query.assert_called_once()
@pytest.mark.vcr(record_mode='none', filter_headers=["authorization"])
@pytest.mark.vcr(record_mode="none", filter_headers=["authorization"])
def test_agent_with_only_crewai_knowledge():
mock_knowledge = MagicMock(spec=Knowledge)
@@ -1928,33 +1931,10 @@ def test_agent_with_only_crewai_knowledge():
role="Information Agent",
goal="Provide information based on knowledge sources",
backstory="You have access to specific knowledge sources.",
llm=LLM(model="openrouter/openai/gpt-4o-mini",api_key=os.getenv('OPENROUTER_API_KEY'))
)
# Create a task that requires the agent to use the knowledge
task = Task(
description="What is Vidit's favorite color?",
expected_output="Vidit's favorclearite color.",
agent=agent
)
crew = Crew(agents=[agent], tasks=[task],knowledge=mock_knowledge)
crew.kickoff()
mock_knowledge.query.assert_called_once()
@pytest.mark.vcr(record_mode='none', filter_headers=["authorization"])
def test_agent_knowledege_with_crewai_knowledge():
crew_knowledge = MagicMock(spec=Knowledge)
agent_knowledge = MagicMock(spec=Knowledge)
agent = Agent(
role="Information Agent",
goal="Provide information based on knowledge sources",
backstory="You have access to specific knowledge sources.",
llm=LLM(model="openrouter/openai/gpt-4o-mini",api_key=os.getenv('OPENROUTER_API_KEY')),
knowledge=agent_knowledge
llm=LLM(
model="openrouter/openai/gpt-4o-mini",
api_key=os.getenv("OPENROUTER_API_KEY"),
),
)
# Create a task that requires the agent to use the knowledge
@@ -1964,7 +1944,35 @@ def test_agent_knowledege_with_crewai_knowledge():
agent=agent,
)
crew = Crew(agents=[agent],tasks=[task],knowledge=crew_knowledge)
crew = Crew(agents=[agent], tasks=[task], knowledge=mock_knowledge)
crew.kickoff()
mock_knowledge.query.assert_called_once()
@pytest.mark.vcr(record_mode="none", filter_headers=["authorization"])
def test_agent_knowledege_with_crewai_knowledge():
crew_knowledge = MagicMock(spec=Knowledge)
agent_knowledge = MagicMock(spec=Knowledge)
agent = Agent(
role="Information Agent",
goal="Provide information based on knowledge sources",
backstory="You have access to specific knowledge sources.",
llm=LLM(
model="openrouter/openai/gpt-4o-mini",
api_key=os.getenv("OPENROUTER_API_KEY"),
),
knowledge=agent_knowledge,
)
# Create a task that requires the agent to use the knowledge
task = Task(
description="What is Vidit's favorite color?",
expected_output="Vidit's favorclearite color.",
agent=agent,
)
crew = Crew(agents=[agent], tasks=[task], knowledge=crew_knowledge)
crew.kickoff()
agent_knowledge.query.assert_called_once()
crew_knowledge.query.assert_called_once()
@@ -2164,7 +2172,12 @@ def mock_get_auth_token():
@patch("crewai.cli.plus_api.PlusAPI.get_agent")
def test_agent_from_repository(mock_get_agent, mock_get_auth_token):
from crewai_tools import SerperDevTool, XMLSearchTool, CSVSearchTool, EnterpriseActionTool
from crewai_tools import (
SerperDevTool,
XMLSearchTool,
CSVSearchTool,
EnterpriseActionTool,
)
mock_get_response = MagicMock()
mock_get_response.status_code = 200
@@ -2173,12 +2186,23 @@ def test_agent_from_repository(mock_get_agent, mock_get_auth_token):
"goal": "test goal",
"backstory": "test backstory",
"tools": [
{"module": "crewai_tools", "name": "SerperDevTool", "init_params": {"n_results": 30}},
{"module": "crewai_tools", "name": "XMLSearchTool", "init_params": {"summarize": True}},
{
"module": "crewai_tools",
"name": "SerperDevTool",
"init_params": {"n_results": "30"},
},
{
"module": "crewai_tools",
"name": "XMLSearchTool",
"init_params": {"summarize": "true"},
},
{"module": "crewai_tools", "name": "CSVSearchTool", "init_params": {}},
# using a tools that returns a list of BaseTools
{"module": "crewai_tools", "name": "CrewaiEnterpriseTools", "init_params": {"actions_list": [], "enterprise_token": "test_key"}},
{
"module": "crewai_tools",
"name": "CrewaiEnterpriseTools",
"init_params": {"actions_list": [], "enterprise_token": "test_key"},
},
],
}
mock_get_agent.return_value = mock_get_response
@@ -2221,7 +2245,9 @@ def test_agent_from_repository_override_attributes(mock_get_agent, mock_get_auth
"role": "test role",
"goal": "test goal",
"backstory": "test backstory",
"tools": [{"name": "SerperDevTool", "module": "crewai_tools", "init_params": {}}],
"tools": [
{"name": "SerperDevTool", "module": "crewai_tools", "init_params": {}}
],
}
mock_get_agent.return_value = mock_get_response
agent = Agent(from_repository="test_agent", role="Custom Role")

File diff suppressed because one or more lines are too long

View File

@@ -2,6 +2,7 @@
import os
import tempfile
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
from dotenv import load_dotenv
@@ -34,14 +35,130 @@ def setup_test_environment():
f"Test storage directory {storage_dir} is not writable: {e}"
)
# Set environment variable to point to the test storage directory
os.environ["CREWAI_STORAGE_DIR"] = str(storage_dir)
os.environ["CREWAI_TESTING"] = "true"
yield
os.environ.pop("CREWAI_TESTING", None)
# Cleanup is handled automatically when tempfile context exits
def pytest_configure(config):
config.addinivalue_line(
"markers", "telemetry: mark test as a telemetry test (don't mock telemetry)"
)
@pytest.fixture(autouse=True)
def auto_mock_telemetry(request):
if request.node.get_closest_marker("telemetry"):
telemetry_env = {
key: value
for key, value in os.environ.items()
if key not in ["CREWAI_DISABLE_TELEMETRY", "OTEL_SDK_DISABLED"]
}
with patch.dict(os.environ, telemetry_env, clear=True):
yield
return
if "telemetry" in str(request.fspath):
telemetry_env = {
key: value
for key, value in os.environ.items()
if key not in ["CREWAI_DISABLE_TELEMETRY", "OTEL_SDK_DISABLED"]
}
with patch.dict(os.environ, telemetry_env, clear=True):
yield
return
with patch.dict(
os.environ, {"CREWAI_DISABLE_TELEMETRY": "true", "OTEL_SDK_DISABLED": "true"}
):
with patch("crewai.telemetry.Telemetry") as mock_telemetry_class:
mock_instance = create_mock_telemetry_instance()
mock_telemetry_class.return_value = mock_instance
with (
patch(
"crewai.utilities.events.event_listener.Telemetry",
mock_telemetry_class,
),
patch("crewai.tools.tool_usage.Telemetry", mock_telemetry_class),
patch("crewai.cli.command.Telemetry", mock_telemetry_class),
patch("crewai.cli.create_flow.Telemetry", mock_telemetry_class),
):
yield mock_instance
def create_mock_telemetry_instance():
mock_instance = Mock()
mock_instance.ready = False
mock_instance.trace_set = False
mock_instance._initialized = True
mock_instance._is_telemetry_disabled.return_value = True
mock_instance._should_execute_telemetry.return_value = False
telemetry_methods = [
"set_tracer",
"crew_creation",
"task_started",
"task_ended",
"tool_usage",
"tool_repeated_usage",
"tool_usage_error",
"crew_execution_span",
"end_crew",
"flow_creation_span",
"flow_execution_span",
"individual_test_result_span",
"test_execution_span",
"deploy_signup_error_span",
"start_deployment_span",
"create_crew_deployment_span",
"get_crew_logs_span",
"remove_crew_span",
"flow_plotting_span",
"_add_attribute",
"_safe_telemetry_operation",
]
for method in telemetry_methods:
setattr(mock_instance, method, Mock(return_value=None))
mock_instance.task_started.return_value = None
return mock_instance
@pytest.fixture
def mock_opentelemetry_components():
with (
patch("opentelemetry.trace.get_tracer") as mock_get_tracer,
patch("opentelemetry.trace.set_tracer_provider") as mock_set_provider,
patch("opentelemetry.baggage.set_baggage") as mock_set_baggage,
patch("opentelemetry.baggage.get_baggage") as mock_get_baggage,
patch("opentelemetry.context.attach") as mock_attach,
patch("opentelemetry.context.detach") as mock_detach,
):
mock_tracer = Mock()
mock_span = Mock()
mock_tracer.start_span.return_value = mock_span
mock_get_tracer.return_value = mock_tracer
yield {
"get_tracer": mock_get_tracer,
"set_tracer_provider": mock_set_provider,
"tracer": mock_tracer,
"span": mock_span,
"set_baggage": mock_set_baggage,
"get_baggage": mock_get_baggage,
"attach": mock_attach,
"detach": mock_detach,
}
@pytest.fixture(scope="module")
def vcr_config(request) -> dict:
return {

View File

@@ -1,5 +1,4 @@
"""Test Agent creation and execution basic functionality."""
import hashlib
import json
from concurrent.futures import Future
@@ -51,7 +50,7 @@ from crewai.utilities.events.memory_events import (
MemoryRetrievalStartedEvent,
MemoryRetrievalCompletedEvent,
)
from crewai.memory.external.external_memory import ExternalMemory
@pytest.fixture
def ceo():
@@ -312,7 +311,6 @@ def test_crew_creation(researcher, writer):
@pytest.mark.vcr(filter_headers=["authorization"])
def test_sync_task_execution(researcher, writer):
from unittest.mock import patch
tasks = [
Task(
@@ -961,7 +959,6 @@ def test_cache_hitting_between_agents(researcher, writer, ceo):
@pytest.mark.vcr(filter_headers=["authorization"])
def test_api_calls_throttling(capsys):
from unittest.mock import patch
from crewai.tools import tool
@@ -1396,7 +1393,6 @@ def test_kickoff_for_each_invalid_input():
def test_kickoff_for_each_error_handling():
"""Tests error handling in kickoff_for_each when kickoff raises an error."""
from unittest.mock import patch
inputs = [
{"topic": "dog"},
@@ -1433,7 +1429,6 @@ def test_kickoff_for_each_error_handling():
@pytest.mark.asyncio
async def test_kickoff_async_basic_functionality_and_output():
"""Tests the basic functionality and output of kickoff_async."""
from unittest.mock import patch
inputs = {"topic": "dog"}
@@ -1540,7 +1535,6 @@ async def test_async_kickoff_for_each_async_empty_input():
def test_set_agents_step_callback():
from unittest.mock import patch
researcher_agent = Agent(
role="Researcher",
@@ -1570,7 +1564,6 @@ def test_set_agents_step_callback():
def test_dont_set_agents_step_callback_if_already_set():
from unittest.mock import patch
def agent_callback(_):
pass
@@ -2035,7 +2028,6 @@ def test_crew_inputs_interpolate_both_agents_and_tasks():
def test_crew_inputs_interpolate_both_agents_and_tasks_diff():
from unittest.mock import patch
agent = Agent(
role="{topic} Researcher",
@@ -2068,7 +2060,6 @@ def test_crew_inputs_interpolate_both_agents_and_tasks_diff():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_crew_does_not_interpolate_without_inputs():
from unittest.mock import patch
agent = Agent(
role="{topic} Researcher",
@@ -2203,7 +2194,6 @@ def test_task_same_callback_both_on_task_and_crew():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_tools_with_custom_caching():
from unittest.mock import patch
from crewai.tools import tool
@@ -2484,7 +2474,6 @@ def test_multiple_conditional_tasks(researcher, writer):
@pytest.mark.vcr(filter_headers=["authorization"])
def test_using_contextual_memory():
from unittest.mock import patch
math_researcher = Agent(
role="Researcher",
@@ -2583,7 +2572,6 @@ def test_memory_events_are_emitted():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_using_contextual_memory_with_long_term_memory():
from unittest.mock import patch
math_researcher = Agent(
role="Researcher",
@@ -2614,7 +2602,6 @@ def test_using_contextual_memory_with_long_term_memory():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_warning_long_term_memory_without_entity_memory():
from unittest.mock import patch
math_researcher = Agent(
role="Researcher",
@@ -2651,7 +2638,6 @@ def test_warning_long_term_memory_without_entity_memory():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_long_term_memory_with_memory_flag():
from unittest.mock import patch
math_researcher = Agent(
role="Researcher",
@@ -2686,7 +2672,6 @@ def test_long_term_memory_with_memory_flag():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_using_contextual_memory_with_short_term_memory():
from unittest.mock import patch
math_researcher = Agent(
role="Researcher",
@@ -2717,7 +2702,6 @@ def test_using_contextual_memory_with_short_term_memory():
@pytest.mark.vcr(filter_headers=["authorization"])
def test_disabled_memory_using_contextual_memory():
from unittest.mock import patch
math_researcher = Agent(
role="Researcher",
@@ -2845,7 +2829,6 @@ def test_crew_output_file_validation_failures():
def test_manager_agent(researcher, writer):
from unittest.mock import patch
task = Task(
description="Come up with a list of 5 interesting ideas to explore for an article, then write one amazing paragraph highlight for each idea that showcases how good an article about this topic could be. Return the list of ideas with their paragraph and your notes.",
@@ -4752,3 +4735,43 @@ def test_default_crew_name(researcher, writer):
],
)
assert crew.name == "crew"
@pytest.mark.vcr(filter_headers=["authorization"])
def test_ensure_exchanged_messages_are_propagated_to_external_memory():
external_memory = ExternalMemory(storage=MagicMock())
math_researcher = Agent(
role="Researcher",
goal="You research about math.",
backstory="You're an expert in research and you love to learn new things.",
allow_delegation=False,
)
task1 = Task(
description="Research a topic to teach a kid aged 6 about math.",
expected_output="A topic, explanation, angle, and examples.",
agent=math_researcher,
)
crew = Crew(
agents=[math_researcher],
tasks=[task1],
external_memory=external_memory,
)
with patch.object(
ExternalMemory, "save", return_value=None
) as external_memory_save:
crew.kickoff()
expected_messages = [
{'role': 'system', 'content': "You are Researcher. You're an expert in research and you love to learn new things.\nYour personal goal is: You research about math.\nTo give my best complete final answer to the task respond using the exact following format:\n\nThought: I now can give a great answer\nFinal Answer: Your final answer must be the great and the most complete as possible, it must be outcome described.\n\nI MUST use these formats, my job depends on it!"},
{'role': 'user', 'content': '\nCurrent Task: Research a topic to teach a kid aged 6 about math.\n\nThis is the expected criteria for your final answer: A topic, explanation, angle, and examples.\nyou MUST return the actual complete content as the final answer, not a summary.\n\n# Useful context: \nExternal memories:\n\n\nBegin! This is VERY important to you, use the tools available and give your best Final Answer, your job depends on it!\n\nThought:'},
{'role': 'assistant', 'content': 'I now can give a great answer \nFinal Answer: \n\n**Topic: Understanding Shapes (Geometry)**\n\n**Explanation:** \nShapes are everywhere around us! They are the special forms that we can see in everyday objects. Teaching a 6-year-old about shapes is not only fun but also a way to help them think about the world around them and develop their spatial awareness. We will focus on basic shapes: circle, square, triangle, and rectangle. Understanding these shapes helps kids recognize and describe their environment.\n\n**Angle:** \nLets make learning about shapes an adventure! We can turn it into a treasure hunt where the child has to find objects around the house or outside that match the shapes we learn. This hands-on approach helps make the learning stick!\n\n**Examples:** \n1. **Circle:** \n - Explanation: A circle is round and has no corners. It looks like a wheel or a cookie! \n - Activity: Find objects that are circles, such as a clock, a dinner plate, or a ball. Draw a big circle on a paper and then try to draw smaller circles inside it.\n\n2. **Square:** \n - Explanation: A square has four equal sides and four corners. It looks like a box! \n - Activity: Look for squares in books, in windows, or in building blocks. Try to build a tall tower using square blocks!\n\n3. **Triangle:** \n - Explanation: A triangle has three sides and three corners. It looks like a slice of pizza or a roof! \n - Activity: Use crayons to draw a big triangle and then find things that are shaped like a triangle, like a slice of cheese or a traffic sign.\n\n4. **Rectangle:** \n - Explanation: A rectangle has four sides but only opposite sides are equal. Its like a stretched square! \n - Activity: Search for rectangles, such as a book cover or a door. You can cut out rectangles from colored paper and create a collage!\n\nBy relating the shapes to fun activities and using real-world examples, we not only make learning more enjoyable but also help the child better remember and understand the concept of shapes in math. This foundation forms the basis of their future learning in geometry!'}
]
external_memory_save.assert_called_once_with(
value=ANY,
metadata={"description": ANY, "messages": expected_messages},
agent=ANY,
)

View File

@@ -39,6 +39,7 @@ def test_short_term_memory_search_events(short_term_memory):
events = defaultdict(list)
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(MemoryQueryStartedEvent)
def on_search_started(source, event):
events["MemoryQueryStartedEvent"].append(event)
@@ -59,33 +60,34 @@ def test_short_term_memory_search_events(short_term_memory):
assert len(events["MemoryQueryFailedEvent"]) == 0
assert dict(events["MemoryQueryStartedEvent"][0]) == {
'timestamp': ANY,
'type': 'memory_query_started',
'source_fingerprint': None,
'source_type': 'short_term_memory',
'fingerprint_metadata': None,
'query': 'test value',
'limit': 3,
'score_threshold': 0.35
"timestamp": ANY,
"type": "memory_query_started",
"source_fingerprint": None,
"source_type": "short_term_memory",
"fingerprint_metadata": None,
"query": "test value",
"limit": 3,
"score_threshold": 0.35,
}
assert dict(events["MemoryQueryCompletedEvent"][0]) == {
'timestamp': ANY,
'type': 'memory_query_completed',
'source_fingerprint': None,
'source_type': 'short_term_memory',
'fingerprint_metadata': None,
'query': 'test value',
'results': [],
'limit': 3,
'score_threshold': 0.35,
'query_time_ms': ANY
"timestamp": ANY,
"type": "memory_query_completed",
"source_fingerprint": None,
"source_type": "short_term_memory",
"fingerprint_metadata": None,
"query": "test value",
"results": [],
"limit": 3,
"score_threshold": 0.35,
"query_time_ms": ANY,
}
def test_short_term_memory_save_events(short_term_memory):
events = defaultdict(list)
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(MemorySaveStartedEvent)
def on_save_started(source, event):
events["MemorySaveStartedEvent"].append(event)
@@ -105,28 +107,29 @@ def test_short_term_memory_save_events(short_term_memory):
assert len(events["MemorySaveFailedEvent"]) == 0
assert dict(events["MemorySaveStartedEvent"][0]) == {
'timestamp': ANY,
'type': 'memory_save_started',
'source_fingerprint': None,
'source_type': 'short_term_memory',
'fingerprint_metadata': None,
'value': 'test value',
'metadata': {'task': 'test_task'},
'agent_role': "test_agent"
"timestamp": ANY,
"type": "memory_save_started",
"source_fingerprint": None,
"source_type": "short_term_memory",
"fingerprint_metadata": None,
"value": "test value",
"metadata": {"task": "test_task"},
"agent_role": "test_agent",
}
assert dict(events["MemorySaveCompletedEvent"][0]) == {
'timestamp': ANY,
'type': 'memory_save_completed',
'source_fingerprint': None,
'source_type': 'short_term_memory',
'fingerprint_metadata': None,
'value': 'test value',
'metadata': {'task': 'test_task', 'agent': 'test_agent'},
'agent_role': "test_agent",
'save_time_ms': ANY
"timestamp": ANY,
"type": "memory_save_completed",
"source_fingerprint": None,
"source_type": "short_term_memory",
"fingerprint_metadata": None,
"value": "test value",
"metadata": {"task": "test_task", "agent": "test_agent"},
"agent_role": "test_agent",
"save_time_ms": ANY,
}
def test_save_and_search(short_term_memory):
memory = ShortTermMemoryItem(
data="""test value test value test value test value test value test value

View File

@@ -1,4 +1,5 @@
import os
import threading
from unittest.mock import patch
import pytest
@@ -11,12 +12,16 @@ from opentelemetry import trace
@pytest.fixture(autouse=True)
def cleanup_telemetry():
"""Automatically clean up Telemetry singleton between tests."""
Telemetry._instance = None
if hasattr(Telemetry, "_lock"):
Telemetry._lock = threading.Lock()
yield
Telemetry._instance = None
if hasattr(Telemetry, "_lock"):
Telemetry._lock = threading.Lock()
@pytest.mark.telemetry
@pytest.mark.parametrize(
"env_var,value,expected_ready",
[
@@ -36,6 +41,7 @@ def test_telemetry_environment_variables(env_var, value, expected_ready):
assert telemetry.ready is expected_ready
@pytest.mark.telemetry
def test_telemetry_enabled_by_default():
"""Test that telemetry is enabled by default."""
with patch.dict(os.environ, {}, clear=True):
@@ -44,6 +50,7 @@ def test_telemetry_enabled_by_default():
assert telemetry.ready is True
@pytest.mark.telemetry
@patch("crewai.telemetry.telemetry.logger.error")
@patch(
"opentelemetry.exporter.otlp.proto.http.trace_exporter.OTLPSpanExporter.export",
@@ -76,6 +83,7 @@ def test_telemetry_fails_due_connect_timeout(export_mock, logger_mock):
logger_mock.assert_called_once_with(error)
@pytest.mark.telemetry
def test_telemetry_singleton_pattern():
"""Test that Telemetry uses the singleton pattern correctly."""
Telemetry._instance = None

View File

@@ -14,14 +14,18 @@ def cleanup_telemetry():
Telemetry._instance = None
@pytest.mark.parametrize("env_var,value,expected_ready", [
("OTEL_SDK_DISABLED", "true", False),
("OTEL_SDK_DISABLED", "TRUE", False),
("CREWAI_DISABLE_TELEMETRY", "true", False),
("CREWAI_DISABLE_TELEMETRY", "TRUE", False),
("OTEL_SDK_DISABLED", "false", True),
("CREWAI_DISABLE_TELEMETRY", "false", True),
])
@pytest.mark.telemetry
@pytest.mark.parametrize(
"env_var,value,expected_ready",
[
("OTEL_SDK_DISABLED", "true", False),
("OTEL_SDK_DISABLED", "TRUE", False),
("CREWAI_DISABLE_TELEMETRY", "true", False),
("CREWAI_DISABLE_TELEMETRY", "TRUE", False),
("OTEL_SDK_DISABLED", "false", True),
("CREWAI_DISABLE_TELEMETRY", "false", True),
],
)
def test_telemetry_environment_variables(env_var, value, expected_ready):
"""Test telemetry state with different environment variable configurations."""
with patch.dict(os.environ, {env_var: value}):
@@ -30,6 +34,7 @@ def test_telemetry_environment_variables(env_var, value, expected_ready):
assert telemetry.ready is expected_ready
@pytest.mark.telemetry
def test_telemetry_enabled_by_default():
"""Test that telemetry is enabled by default."""
with patch.dict(os.environ, {}, clear=True):
@@ -38,57 +43,60 @@ def test_telemetry_enabled_by_default():
assert telemetry.ready is True
@pytest.mark.telemetry
def test_telemetry_disable_after_singleton_creation():
"""Test that telemetry operations are disabled when env var is set after singleton creation."""
with patch.dict(os.environ, {}, clear=True):
with patch("crewai.telemetry.telemetry.TracerProvider"):
telemetry = Telemetry()
assert telemetry.ready is True
mock_operation = MagicMock()
telemetry._safe_telemetry_operation(mock_operation)
mock_operation.assert_called_once()
mock_operation.reset_mock()
os.environ['CREWAI_DISABLE_TELEMETRY'] = 'true'
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"
telemetry._safe_telemetry_operation(mock_operation)
mock_operation.assert_not_called()
@pytest.mark.telemetry
def test_telemetry_disable_with_multiple_instances():
"""Test that multiple telemetry instances respect dynamically changed env vars."""
with patch.dict(os.environ, {}, clear=True):
with patch("crewai.telemetry.telemetry.TracerProvider"):
telemetry1 = Telemetry()
assert telemetry1.ready is True
os.environ['CREWAI_DISABLE_TELEMETRY'] = 'true'
os.environ["CREWAI_DISABLE_TELEMETRY"] = "true"
telemetry2 = Telemetry()
assert telemetry2 is telemetry1
assert telemetry2.ready is True
mock_operation = MagicMock()
telemetry2._safe_telemetry_operation(mock_operation)
mock_operation.assert_not_called()
@pytest.mark.telemetry
def test_telemetry_otel_sdk_disabled_after_creation():
"""Test that OTEL_SDK_DISABLED also works when set after singleton creation."""
with patch.dict(os.environ, {}, clear=True):
with patch("crewai.telemetry.telemetry.TracerProvider"):
telemetry = Telemetry()
assert telemetry.ready is True
mock_operation = MagicMock()
telemetry._safe_telemetry_operation(mock_operation)
mock_operation.assert_called_once()
mock_operation.reset_mock()
os.environ['OTEL_SDK_DISABLED'] = 'true'
os.environ["OTEL_SDK_DISABLED"] = "true"
telemetry._safe_telemetry_operation(mock_operation)
mock_operation.assert_not_called()

View File

@@ -2,8 +2,8 @@ import os
import pytest
from unittest.mock import patch, MagicMock
# Remove the module-level patch
from crewai import Agent, Task, Crew
from crewai.flow.flow import Flow, start
from crewai.utilities.events.listeners.tracing.trace_listener import (
TraceCollectionListener,
)
@@ -284,29 +284,42 @@ class TestTraceListenerSetup:
f"Found {len(trace_handlers)} trace handlers when tracing should be disabled"
)
def test_trace_listener_setup_correctly(self):
def test_trace_listener_setup_correctly_for_crew(self):
"""Test that trace listener is set up correctly when enabled"""
with patch.dict(os.environ, {"CREWAI_TRACING_ENABLED": "true"}):
trace_listener = TraceCollectionListener()
agent = Agent(
role="Test Agent",
goal="Test goal",
backstory="Test backstory",
llm="gpt-4o-mini",
)
task = Task(
description="Say hello to the world",
expected_output="hello world",
agent=agent,
)
with patch.object(
TraceCollectionListener, "setup_listeners"
) as mock_listener_setup:
Crew(agents=[agent], tasks=[task], verbose=True)
assert mock_listener_setup.call_count >= 1
assert trace_listener.trace_enabled is True
assert trace_listener.batch_manager is not None
@pytest.mark.vcr(filter_headers=["authorization"])
def test_trace_listener_setup_correctly_with_tracing_flag(self):
def test_trace_listener_setup_correctly_for_flow(self):
"""Test that trace listener is set up correctly when enabled"""
agent = Agent(role="Test Agent", goal="Test goal", backstory="Test backstory")
task = Task(
description="Say hello to the world",
expected_output="hello world",
agent=agent,
)
crew = Crew(agents=[agent], tasks=[task], verbose=True, tracing=True)
crew.kickoff()
trace_listener = TraceCollectionListener(tracing=True)
assert trace_listener.trace_enabled is True
assert trace_listener.batch_manager is not None
with patch.dict(os.environ, {"CREWAI_TRACING_ENABLED": "true"}):
class FlowExample(Flow):
@start()
def start(self):
pass
with patch.object(
TraceCollectionListener, "setup_listeners"
) as mock_listener_setup:
FlowExample()
assert mock_listener_setup.call_count >= 1
# Helper method to ensure cleanup
def teardown_method(self):

View File

@@ -0,0 +1 @@

View File

@@ -0,0 +1,196 @@
from unittest.mock import Mock, patch
from crewai.utilities.events.third_party.agentops_listener import AgentOpsListener
from crewai.utilities.events.crew_events import (
CrewKickoffStartedEvent,
CrewKickoffCompletedEvent,
)
from crewai.utilities.events.task_events import TaskEvaluationEvent
from crewai.utilities.events.tool_usage_events import (
ToolUsageStartedEvent,
ToolUsageErrorEvent,
)
from crewai.utilities.events.crewai_event_bus import CrewAIEventsBus
class TestAgentOpsListener:
def test_agentops_listener_initialization_with_agentops_installed(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops"):
listener = AgentOpsListener()
assert listener.agentops is not None
def test_agentops_listener_initialization_without_agentops_installed(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops", side_effect=ImportError):
listener = AgentOpsListener()
assert listener.agentops is None
def test_setup_listeners_with_agentops_installed(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops"):
listener = AgentOpsListener()
mock_event_bus = Mock(spec=CrewAIEventsBus)
listener.setup_listeners(mock_event_bus)
assert mock_event_bus.register_handler.call_count == 5
mock_event_bus.register_handler.assert_any_call(
CrewKickoffStartedEvent, listener._handle_crew_kickoff_started
)
mock_event_bus.register_handler.assert_any_call(
CrewKickoffCompletedEvent, listener._handle_crew_kickoff_completed
)
mock_event_bus.register_handler.assert_any_call(
ToolUsageStartedEvent, listener._handle_tool_usage_started
)
mock_event_bus.register_handler.assert_any_call(
ToolUsageErrorEvent, listener._handle_tool_usage_error
)
mock_event_bus.register_handler.assert_any_call(
TaskEvaluationEvent, listener._handle_task_evaluation
)
def test_setup_listeners_without_agentops_installed(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops", side_effect=ImportError):
listener = AgentOpsListener()
mock_event_bus = Mock(spec=CrewAIEventsBus)
listener.setup_listeners(mock_event_bus)
mock_event_bus.register_handler.assert_not_called()
def test_handle_crew_kickoff_started_with_agentops(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
listener = AgentOpsListener()
event = CrewKickoffStartedEvent(crew_id="test-crew")
listener._handle_crew_kickoff_started(event)
mock_agentops.start_session.assert_called_once()
call_args = mock_agentops.start_session.call_args
assert call_args[1]["tags"] == ["crewai", "crew_kickoff"]
def test_handle_crew_kickoff_started_without_agentops(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops", side_effect=ImportError):
listener = AgentOpsListener()
event = CrewKickoffStartedEvent(crew_id="test-crew")
listener._handle_crew_kickoff_started(event)
def test_handle_crew_kickoff_completed_with_agentops(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
listener = AgentOpsListener()
event = CrewKickoffCompletedEvent(crew_id="test-crew", crew_output=Mock())
listener._handle_crew_kickoff_completed(event)
mock_agentops.end_session.assert_called_once_with("Success")
def test_handle_crew_kickoff_completed_without_agentops(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops", side_effect=ImportError):
listener = AgentOpsListener()
event = CrewKickoffCompletedEvent(crew_id="test-crew", crew_output=Mock())
listener._handle_crew_kickoff_completed(event)
def test_handle_tool_usage_started_with_agentops(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
listener = AgentOpsListener()
event = ToolUsageStartedEvent(
tool_name="test_tool",
arguments={"arg1": "value1"},
agent_id="test-agent",
task_id="test-task"
)
listener._handle_tool_usage_started(event)
mock_agentops.record.assert_called_once()
call_args = mock_agentops.record.call_args[0][0]
assert hasattr(call_args, "action_type")
def test_handle_tool_usage_error_with_agentops(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
listener = AgentOpsListener()
event = ToolUsageErrorEvent(
tool_name="test_tool",
arguments={"arg1": "value1"},
error="Test error",
agent_id="test-agent",
task_id="test-task"
)
listener._handle_tool_usage_error(event)
mock_agentops.record.assert_called_once()
def test_handle_task_evaluation_with_agentops(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
listener = AgentOpsListener()
event = TaskEvaluationEvent(
task_id="test-task",
score=0.85,
feedback="Good performance"
)
listener._handle_task_evaluation(event)
mock_agentops.record.assert_called_once()
def test_handle_crew_kickoff_started_with_exception(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
mock_agentops.start_session.side_effect = Exception("Test exception")
listener = AgentOpsListener()
event = CrewKickoffStartedEvent(crew_id="test-crew")
listener._handle_crew_kickoff_started(event)
def test_handle_crew_kickoff_completed_with_exception(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
mock_agentops.end_session.side_effect = Exception("Test exception")
listener = AgentOpsListener()
event = CrewKickoffCompletedEvent(crew_id="test-crew", crew_output=Mock())
listener._handle_crew_kickoff_completed(event)
def test_handle_tool_usage_started_with_exception(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
mock_agentops.record.side_effect = Exception("Test exception")
listener = AgentOpsListener()
event = ToolUsageStartedEvent(
tool_name="test_tool",
arguments={"arg1": "value1"},
agent_id="test-agent",
task_id="test-task"
)
listener._handle_tool_usage_started(event)
def test_handle_tool_usage_error_with_exception(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
mock_agentops.record.side_effect = Exception("Test exception")
listener = AgentOpsListener()
event = ToolUsageErrorEvent(
tool_name="test_tool",
arguments={"arg1": "value1"},
error="Test error",
agent_id="test-agent",
task_id="test-task"
)
listener._handle_tool_usage_error(event)
def test_handle_task_evaluation_with_exception(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops") as mock_agentops:
mock_agentops.record.side_effect = Exception("Test exception")
listener = AgentOpsListener()
event = TaskEvaluationEvent(
task_id="test-task",
score=0.85,
feedback="Good performance"
)
listener._handle_task_evaluation(event)
def test_agentops_listener_instance_creation(self):
with patch("crewai.utilities.events.third_party.agentops_listener.agentops"):
from crewai.utilities.events.third_party.agentops_listener import agentops_listener
assert agentops_listener is not None
assert isinstance(agentops_listener, AgentOpsListener)

45
uv.lock generated
View File

@@ -34,22 +34,6 @@ resolution-markers = [
"(python_full_version < '3.11' and platform_machine != 'aarch64' and platform_python_implementation == 'PyPy' and sys_platform == 'linux') or (python_full_version < '3.11' and platform_python_implementation == 'PyPy' and sys_platform != 'darwin' and sys_platform != 'linux')",
]
[[package]]
name = "agentops"
version = "0.3.18"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "packaging" },
{ name = "psutil" },
{ name = "pyyaml" },
{ name = "requests" },
{ name = "termcolor" },
]
sdist = { url = "https://files.pythonhosted.org/packages/c5/52/46bb2f29b9e5f2e1d8b124296b7794934a9048de635d9e7d6a95e791ad7b/agentops-0.3.18.tar.gz", hash = "sha256:4d509754df7be52579597cc9f53939c5218131a0379463e0ff6f6f40cde9fcc4", size = 55394, upload-time = "2024-11-19T19:06:21.306Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/97/8d/bd4cad95dad722dc2d3e4179feab1058ef846828c0e15e51e8bfaea373ee/agentops-0.3.18-py3-none-any.whl", hash = "sha256:8b48d8a1662f276653430fd541c77fa4f9a15a43e881b518ff88ea56925afcf7", size = 58032, upload-time = "2024-11-19T19:06:19.068Z" },
]
[[package]]
name = "aiohappyeyeballs"
version = "2.6.1"
@@ -744,9 +728,6 @@ dependencies = [
]
[package.optional-dependencies]
agentops = [
{ name = "agentops" },
]
aisuite = [
{ name = "aisuite" },
]
@@ -792,7 +773,6 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "agentops", marker = "extra == 'agentops'", specifier = "==0.3.18" },
{ name = "aisuite", marker = "extra == 'aisuite'", specifier = ">=0.1.10" },
{ name = "appdirs", specifier = ">=1.4.4" },
{ name = "blinker", specifier = ">=1.9.0" },
@@ -828,7 +808,7 @@ requires-dist = [
{ name = "tomli-w", specifier = ">=1.1.0" },
{ name = "uv", specifier = ">=0.4.25" },
]
provides-extras = ["agentops", "aisuite", "docling", "embeddings", "mem0", "openpyxl", "pandas", "pdfplumber", "tools"]
provides-extras = ["aisuite", "docling", "embeddings", "mem0", "openpyxl", "pandas", "pdfplumber", "tools"]
[package.metadata.requires-dev]
dev = [
@@ -4013,20 +3993,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/7e/cc/7e77861000a0691aeea8f4566e5d3aa716f2b1dece4a24439437e41d3d25/protobuf-5.29.5-py3-none-any.whl", hash = "sha256:6cf42630262c59b2d8de33954443d94b746c952b01434fc58a417fdbd2e84bd5", size = 172823, upload-time = "2025-05-28T23:51:58.157Z" },
]
[[package]]
name = "psutil"
version = "5.9.8"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/90/c7/6dc0a455d111f68ee43f27793971cf03fe29b6ef972042549db29eec39a2/psutil-5.9.8.tar.gz", hash = "sha256:6be126e3225486dff286a8fb9a06246a5253f4c7c53b475ea5f5ac934e64194c", size = 503247, upload-time = "2024-01-19T20:47:09.517Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e7/e3/07ae864a636d70a8a6f58da27cb1179192f1140d5d1da10886ade9405797/psutil-5.9.8-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:aee678c8720623dc456fa20659af736241f575d79429a0e5e9cf88ae0605cc81", size = 248702, upload-time = "2024-01-19T20:47:36.303Z" },
{ url = "https://files.pythonhosted.org/packages/b3/bd/28c5f553667116b2598b9cc55908ec435cb7f77a34f2bff3e3ca765b0f78/psutil-5.9.8-cp36-abi3-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:8cb6403ce6d8e047495a701dc7c5bd788add903f8986d523e3e20b98b733e421", size = 285242, upload-time = "2024-01-19T20:47:39.65Z" },
{ url = "https://files.pythonhosted.org/packages/c5/4f/0e22aaa246f96d6ac87fe5ebb9c5a693fbe8877f537a1022527c47ca43c5/psutil-5.9.8-cp36-abi3-manylinux_2_12_x86_64.manylinux2010_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d06016f7f8625a1825ba3732081d77c94589dca78b7a3fc072194851e88461a4", size = 288191, upload-time = "2024-01-19T20:47:43.078Z" },
{ url = "https://files.pythonhosted.org/packages/6e/f5/2aa3a4acdc1e5940b59d421742356f133185667dd190b166dbcfcf5d7b43/psutil-5.9.8-cp37-abi3-win32.whl", hash = "sha256:bc56c2a1b0d15aa3eaa5a60c9f3f8e3e565303b465dbf57a1b730e7a2b9844e0", size = 251252, upload-time = "2024-01-19T20:47:52.88Z" },
{ url = "https://files.pythonhosted.org/packages/93/52/3e39d26feae7df0aa0fd510b14012c3678b36ed068f7d78b8d8784d61f0e/psutil-5.9.8-cp37-abi3-win_amd64.whl", hash = "sha256:8db4c1b57507eef143a15a6884ca10f7c73876cdf5d51e713151c1236a0e68cf", size = 255090, upload-time = "2024-01-19T20:47:56.019Z" },
{ url = "https://files.pythonhosted.org/packages/05/33/2d74d588408caedd065c2497bdb5ef83ce6082db01289a1e1147f6639802/psutil-5.9.8-cp38-abi3-macosx_11_0_arm64.whl", hash = "sha256:d16bbddf0693323b8c6123dd804100241da461e41d6e332fb0ba6058f630f8c8", size = 249898, upload-time = "2024-01-19T20:47:59.238Z" },
]
[[package]]
name = "ptyprocess"
version = "0.7.0"
@@ -5502,15 +5468,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e5/30/643397144bfbfec6f6ef821f36f33e57d35946c44a2352d3c9f0ae847619/tenacity-9.1.2-py3-none-any.whl", hash = "sha256:f77bf36710d8b73a50b2dd155c97b870017ad21afe6ab300326b0371b3b05138", size = 28248, upload-time = "2025-04-02T08:25:07.678Z" },
]
[[package]]
name = "termcolor"
version = "2.4.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/10/56/d7d66a84f96d804155f6ff2873d065368b25a07222a6fd51c4f24ef6d764/termcolor-2.4.0.tar.gz", hash = "sha256:aab9e56047c8ac41ed798fa36d892a37aca6b3e9159f3e0c24bc64a9b3ac7b7a", size = 12664, upload-time = "2023-12-01T11:04:51.66Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d9/5f/8c716e47b3a50cbd7c146f45881e11d9414def768b7cd9c5e6650ec2a80a/termcolor-2.4.0-py3-none-any.whl", hash = "sha256:9297c0df9c99445c2412e832e882a7884038a25617c60cea2ad69488d4040d63", size = 7719, upload-time = "2023-12-01T11:04:50.019Z" },
]
[[package]]
name = "tifffile"
version = "2025.5.10"