mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-07-13 10:55:08 +00:00
Compare commits
10 Commits
docs/deplo
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
88562df973 | ||
|
|
fb8e93be25 | ||
|
|
4fdb7f2bfb | ||
|
|
bfa652a7be | ||
|
|
b65c8487d2 | ||
|
|
a8b3ecb723 | ||
|
|
7967b19057 | ||
|
|
85c467dfe2 | ||
|
|
7baf8f9ba1 | ||
|
|
860817cbcd |
@@ -144,6 +144,18 @@ In this section, you'll find detailed examples that help you select, configure,
|
||||
)
|
||||
```
|
||||
|
||||
**Custom OpenAI-Compatible Endpoint:**
|
||||
```python Code
|
||||
from crewai import LLM
|
||||
|
||||
llm = LLM(
|
||||
model="anthropic/claude-sonnet-4-6",
|
||||
custom_openai=True,
|
||||
base_url="https://your-gateway.example.com/v1",
|
||||
api_key="your-gateway-api-key",
|
||||
)
|
||||
```
|
||||
|
||||
**Advanced Configuration:**
|
||||
```python Code
|
||||
from crewai import LLM
|
||||
|
||||
@@ -240,14 +240,15 @@ from crewai import LLM
|
||||
|
||||
# After (OpenAI-compatible mode, no LiteLLM needed):
|
||||
llm = LLM(
|
||||
model="openai/llama3",
|
||||
model="llama3",
|
||||
custom_openai=True,
|
||||
base_url="http://localhost:11434/v1",
|
||||
api_key="ollama" # Ollama doesn't require a real API key
|
||||
)
|
||||
```
|
||||
|
||||
<Tip>
|
||||
Many local inference servers (Ollama, vLLM, LM Studio, llama.cpp) expose an OpenAI-compatible API. You can use the `openai/` prefix with a custom `base_url` to connect to any of them natively.
|
||||
Many local inference servers (Ollama, vLLM, LM Studio, llama.cpp) expose an OpenAI-compatible API. You can use `custom_openai=True` with a custom `base_url` to connect to any of them natively while keeping the model ID your gateway expects.
|
||||
</Tip>
|
||||
|
||||
### Step 4: Update your YAML configs
|
||||
@@ -295,6 +296,92 @@ crewai run
|
||||
uv run pytest
|
||||
```
|
||||
|
||||
## Custom OpenAI-Compatible Endpoints
|
||||
|
||||
Many providers and local servers (Ollama, vLLM, LM Studio, llama.cpp, LiteLLM proxies, and hosted gateways) expose an **OpenAI-compatible** API. Instead of routing these through LiteLLM, you can talk to them directly with CrewAI's native OpenAI integration by setting `custom_openai=True`.
|
||||
|
||||
This is the recommended replacement for any LiteLLM provider that offers an OpenAI-compatible endpoint.
|
||||
|
||||
### How it works
|
||||
|
||||
- `custom_openai=True` forces CrewAI to use the native OpenAI SDK, regardless of the model name.
|
||||
- The model ID is passed to the endpoint without validation against OpenAI's known-model list. This lets you use arbitrary model IDs your gateway expects (for example, `anthropic/claude-sonnet-4-6` served behind an OpenAI-compatible proxy). An optional leading `openai/` routing prefix is stripped.
|
||||
- A base URL is **required**. CrewAI resolves it, in order, from:
|
||||
1. `base_url=...`
|
||||
2. `api_base=...`
|
||||
3. `OPENAI_BASE_URL` environment variable
|
||||
4. `OPENAI_API_BASE` environment variable (legacy)
|
||||
|
||||
If none are set, CrewAI raises a `ValueError` so misconfiguration fails fast instead of silently hitting `api.openai.com`.
|
||||
|
||||
```python
|
||||
from crewai import LLM
|
||||
|
||||
llm = LLM(
|
||||
model="anthropic/claude-sonnet-4-6", # passed through as-is
|
||||
custom_openai=True,
|
||||
base_url="https://your-gateway.example/v1",
|
||||
api_key="your-key",
|
||||
)
|
||||
```
|
||||
|
||||
### Connect to common servers
|
||||
|
||||
<Tabs>
|
||||
<Tab title="Ollama">
|
||||
```python
|
||||
from crewai import LLM
|
||||
|
||||
llm = LLM(
|
||||
model="llama3.2:latest",
|
||||
custom_openai=True,
|
||||
base_url="http://localhost:11434/v1",
|
||||
api_key="ollama", # Ollama ignores it, but the client requires a value
|
||||
)
|
||||
```
|
||||
</Tab>
|
||||
<Tab title="vLLM">
|
||||
```python
|
||||
from crewai import LLM
|
||||
|
||||
llm = LLM(
|
||||
model="meta-llama/Meta-Llama-3.1-8B-Instruct",
|
||||
custom_openai=True,
|
||||
base_url="http://localhost:8000/v1",
|
||||
api_key="not-needed",
|
||||
)
|
||||
```
|
||||
</Tab>
|
||||
<Tab title="LM Studio">
|
||||
```python
|
||||
from crewai import LLM
|
||||
|
||||
llm = LLM(
|
||||
model="your-loaded-model",
|
||||
custom_openai=True,
|
||||
base_url="http://localhost:1234/v1",
|
||||
api_key="lm-studio",
|
||||
)
|
||||
```
|
||||
</Tab>
|
||||
<Tab title="Env vars">
|
||||
```bash
|
||||
export OPENAI_BASE_URL="https://your-gateway.example/v1"
|
||||
export OPENAI_API_KEY="your-key"
|
||||
```
|
||||
```python
|
||||
from crewai import LLM
|
||||
|
||||
# base_url is picked up from OPENAI_BASE_URL / OPENAI_API_BASE
|
||||
llm = LLM(model="anthropic/claude-sonnet-4-6", custom_openai=True)
|
||||
```
|
||||
</Tab>
|
||||
</Tabs>
|
||||
|
||||
<Tip>
|
||||
If you use the `openai/` prefix with a model that isn't a known OpenAI model and pass `base_url` or `api_base` directly, CrewAI automatically treats it as a custom OpenAI-compatible endpoint. Environment variables alone do not enable automatic routing for unknown models; set `custom_openai=True` when configuring the endpoint through `OPENAI_BASE_URL` or `OPENAI_API_BASE`.
|
||||
</Tip>
|
||||
|
||||
## Quick Reference: Model String Mapping
|
||||
|
||||
Here are common migration paths from LiteLLM-dependent providers to native ones:
|
||||
@@ -321,7 +408,8 @@ llm = LLM(model="anthropic/claude-sonnet-4-20250514") # High quality
|
||||
# Ollama → OpenAI-compatible (keep using local models)
|
||||
# llm = LLM(model="ollama/llama3")
|
||||
llm = LLM(
|
||||
model="openai/llama3",
|
||||
model="llama3",
|
||||
custom_openai=True,
|
||||
base_url="http://localhost:11434/v1",
|
||||
api_key="ollama"
|
||||
)
|
||||
@@ -349,6 +437,9 @@ llm = LLM(
|
||||
<Accordion title="What about environment variables like OPENAI_API_KEY?">
|
||||
Native providers use the same environment variables you're already familiar with. No changes needed for `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMINI_API_KEY`, etc.
|
||||
</Accordion>
|
||||
<Accordion title="How do I connect to Groq, Together AI, or other OpenAI-compatible providers without LiteLLM?">
|
||||
Most of these providers expose an OpenAI-compatible API. Use `custom_openai=True` with their base URL and API key — see [Custom OpenAI-Compatible Endpoints](#custom-openai-compatible-endpoints). For example, Groq: `LLM(model="llama-3.1-70b-versatile", custom_openai=True, base_url="https://api.groq.com/openai/v1", api_key="...")`. The model ID is passed through untouched, so use whatever ID the provider expects.
|
||||
</Accordion>
|
||||
</AccordionGroup>
|
||||
|
||||
## Related Resources
|
||||
|
||||
@@ -568,12 +568,32 @@ FooterKey .footer-key--key {
|
||||
self._default_inputs: dict[str, Any] | None = None
|
||||
self._crew_result: Any = None
|
||||
self._crew_json_path: Any = None
|
||||
# Declarative-flow execution state. A flow renders per-method "STEPS"
|
||||
# (built from flow method events) instead of the crew task list.
|
||||
self._flow_inputs: dict[str, Any] | None = None
|
||||
self._flow_method_types: dict[str, str] = {}
|
||||
self._flow_steps: list[dict[str, Any]] = []
|
||||
self._current_method: str | None = None
|
||||
self._elapsed_frozen: float | None = None
|
||||
self._want_deploy: bool = False
|
||||
self._trace_url: str | None = None
|
||||
self._consent_screen: TraceConsentScreen | None = None
|
||||
self._telemetry: Telemetry | None = None
|
||||
|
||||
@property
|
||||
def _is_flow_run(self) -> bool:
|
||||
"""True for a non-conversational declarative flow (the STEPS view).
|
||||
|
||||
Gates every flow-specific rendering branch so crew and conversational
|
||||
paths stay byte-identical.
|
||||
"""
|
||||
return self._flow is not None and not self._is_conversational
|
||||
|
||||
@property
|
||||
def _run_noun(self) -> str:
|
||||
"""User-facing noun for the run — 'flow' for a declarative flow, else 'crew'."""
|
||||
return "flow" if self._is_flow_run else "crew"
|
||||
|
||||
# ── Layout ──────────────────────────────────────────────
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
@@ -602,6 +622,8 @@ FooterKey .footer-key--key {
|
||||
self._tick_timer = self.set_interval(1 / 8, self._tick)
|
||||
if self._is_conversational and self._flow:
|
||||
self._start_conversational_session()
|
||||
elif self._flow:
|
||||
self._run_flow_worker()
|
||||
elif self._crew:
|
||||
self._run_crew_worker()
|
||||
elif self._crew_json_path:
|
||||
@@ -681,6 +703,49 @@ FooterKey .footer-key--key {
|
||||
except Exception as e:
|
||||
self.call_from_thread(self._on_crew_failed, str(e))
|
||||
|
||||
@work(thread=True, exclusive=True, group="flow")
|
||||
def _run_flow_worker(self) -> None:
|
||||
from crewai.events.listeners.tracing.utils import (
|
||||
set_suppress_tracing_messages,
|
||||
set_tui_mode,
|
||||
)
|
||||
|
||||
set_tui_mode(True)
|
||||
set_suppress_tracing_messages(True)
|
||||
try:
|
||||
# A declarative flow returns either a CrewOutput (has ``.raw``) or a
|
||||
# bare value (str/dict/pydantic); _stringify_output handles both.
|
||||
result = self._flow.kickoff(inputs=self._flow_inputs)
|
||||
output = self._stringify_output(result)
|
||||
with self._lock:
|
||||
self._crew_result = result
|
||||
self.call_from_thread(self._on_crew_done, output)
|
||||
except Exception as e:
|
||||
self.call_from_thread(self._on_crew_failed, str(e))
|
||||
|
||||
def _set_flow_step_status(self, name: str, status: str) -> None:
|
||||
"""Update a flow method step's status. Caller must hold ``self._lock``."""
|
||||
for step in self._flow_steps:
|
||||
if step["name"] == name:
|
||||
step["status"] = status
|
||||
return
|
||||
|
||||
def _clear_current_method(self, finished_name: str) -> None:
|
||||
"""Drop the header's active method once it ends. Caller holds the lock.
|
||||
|
||||
Falls back to another still-active step (methods can overlap) so the
|
||||
header never keeps spinning a method the STEPS list already shows as
|
||||
done or failed.
|
||||
"""
|
||||
if self._current_method != finished_name:
|
||||
return
|
||||
self._current_method = next(
|
||||
(s["name"] for s in self._flow_steps if s["status"] == "active"), None
|
||||
)
|
||||
# The active method changed; drop its agent so the header doesn't show a
|
||||
# stale agent until the next method's agent event arrives.
|
||||
self._current_agent = ""
|
||||
|
||||
def _on_crew_done(self, output: str | None) -> None:
|
||||
with self._lock:
|
||||
self._status = "completed"
|
||||
@@ -694,13 +759,18 @@ FooterKey .footer-key--key {
|
||||
for k in self._task_statuses:
|
||||
if self._task_statuses[k] == "active":
|
||||
self._task_statuses[k] = "done"
|
||||
for step in self._flow_steps:
|
||||
if step["status"] == "active":
|
||||
step["status"] = "done"
|
||||
now = time.time()
|
||||
for entry in self._log_entries:
|
||||
if entry["status"] == "running":
|
||||
if entry["tool_name"] == "memory_save":
|
||||
continue
|
||||
entry["status"] = "timeout"
|
||||
entry["error"] = "No result received before crew completed"
|
||||
entry["error"] = (
|
||||
f"No result received before {self._run_noun} completed"
|
||||
)
|
||||
entry["duration"] = now - entry["start_time"]
|
||||
try:
|
||||
from crewai.events.listeners.tracing.trace_listener import (
|
||||
@@ -739,13 +809,18 @@ FooterKey .footer-key--key {
|
||||
self._is_streaming = False
|
||||
self._current_step = None
|
||||
self._elapsed_frozen = time.time() - self._start_time
|
||||
for step in self._flow_steps:
|
||||
if step["status"] == "active":
|
||||
step["status"] = "failed"
|
||||
now = time.time()
|
||||
for entry in self._log_entries:
|
||||
if entry["status"] == "running":
|
||||
if entry["tool_name"] == "memory_save":
|
||||
continue
|
||||
entry["status"] = "error"
|
||||
entry["error"] = "No result received before crew failed"
|
||||
entry["error"] = (
|
||||
f"No result received before {self._run_noun} failed"
|
||||
)
|
||||
entry["duration"] = now - entry["start_time"]
|
||||
self._tick()
|
||||
self.call_later(self._focus_activity_log)
|
||||
@@ -1156,6 +1231,45 @@ FooterKey .footer-key--key {
|
||||
widget.update(t)
|
||||
return
|
||||
|
||||
if self._is_flow_run:
|
||||
t.append(" STEPS\n", style=f"bold {_C_PRIMARY}")
|
||||
t.append("\n")
|
||||
if not self._flow_steps:
|
||||
t.append(" ○ waiting…\n", style=_C_DIM)
|
||||
for step in self._flow_steps:
|
||||
name = step["name"]
|
||||
max_name = sidebar_width - 6
|
||||
if len(name) > max_name:
|
||||
name = name[: max_name - 1] + "…"
|
||||
status = step.get("status", "pending")
|
||||
if status == "done":
|
||||
t.append(" ✔ ", style=_C_GREEN)
|
||||
t.append(name, style=_C_DIM)
|
||||
elif status == "active":
|
||||
t.append(f" {self._spinner()} ", style=_C_PRIMARY)
|
||||
t.append(name, style=f"bold {_C_TEXT}")
|
||||
elif status == "failed":
|
||||
t.append(" ✘ ", style=_C_RED)
|
||||
t.append(name, style=_C_RED)
|
||||
elif status == "paused":
|
||||
t.append(" ⏸ ", style=_C_TEAL)
|
||||
t.append(name, style=_C_TEAL)
|
||||
else:
|
||||
t.append(" ○ ", style=_C_DIM)
|
||||
t.append(name, style=_C_DIM)
|
||||
if step.get("call_type"):
|
||||
t.append(f" ({step['call_type']})", style=_C_DIM)
|
||||
t.append("\n")
|
||||
|
||||
t.append("\n")
|
||||
t.append(" TOKENS\n", style=f"bold {_C_PRIMARY}")
|
||||
t.append("\n")
|
||||
out = self._output_tokens + self._live_out_tokens
|
||||
t.append(f" ↑ {self._input_tokens:,}\n", style=_C_DIM)
|
||||
t.append(f" ↓ {out:,}\n", style=_C_DIM)
|
||||
widget.update(t)
|
||||
return
|
||||
|
||||
t.append(" TASKS\n", style=f"bold {_C_PRIMARY}")
|
||||
t.append("\n")
|
||||
|
||||
@@ -1225,6 +1339,55 @@ FooterKey .footer-key--key {
|
||||
widget.update(t)
|
||||
return
|
||||
|
||||
if self._is_flow_run:
|
||||
if self._status == "completed":
|
||||
elapsed = self._elapsed_frozen or (time.time() - self._start_time)
|
||||
t.append("✔ ", style=f"bold {_C_GREEN}")
|
||||
t.append("Flow complete", style=f"bold {_C_GREEN}")
|
||||
t.append(f" {elapsed:.1f}s", style=_C_DIM)
|
||||
out = self._output_tokens + self._live_out_tokens
|
||||
parts = []
|
||||
if self._input_tokens:
|
||||
parts.append(f"↑{self._input_tokens:,}")
|
||||
if out:
|
||||
parts.append(f"↓{out:,}")
|
||||
if parts:
|
||||
t.append(f" {' '.join(parts)} tokens", style=_C_DIM)
|
||||
elif self._status == "failed":
|
||||
t.append("✘ ", style=f"bold {_C_RED}")
|
||||
t.append("Failed", style=f"bold {_C_RED}")
|
||||
if self._error:
|
||||
t.append(f"\n{self._error[:120]}", style=_C_RED)
|
||||
elif self._current_method:
|
||||
paused = any(
|
||||
s["name"] == self._current_method and s["status"] == "paused"
|
||||
for s in self._flow_steps
|
||||
)
|
||||
if paused:
|
||||
t.append("⏸ ", style=_C_TEAL)
|
||||
t.append(self._current_method, style=f"bold {_C_TEAL}")
|
||||
else:
|
||||
t.append(f"{self._spinner()} ", style=_C_PRIMARY)
|
||||
t.append(self._current_method, style=f"bold {_C_PRIMARY}")
|
||||
call_type = self._flow_method_types.get(self._current_method)
|
||||
if call_type:
|
||||
t.append(f" ({call_type})", style=_C_DIM)
|
||||
if paused:
|
||||
t.append(" waiting for feedback", style=_C_DIM)
|
||||
elif self._current_agent:
|
||||
t.append("\nAgent: ", style=_C_DIM)
|
||||
t.append(self._current_agent, style=f"bold {_C_TEXT}")
|
||||
else:
|
||||
t.append(f"{self._spinner()} ", style=_C_PRIMARY)
|
||||
# "Working…" once a step has run (between/after methods);
|
||||
# "Starting flow…" only before the first method.
|
||||
t.append(
|
||||
"Working…" if self._flow_steps else "Starting flow…",
|
||||
style=_C_DIM,
|
||||
)
|
||||
widget.update(t)
|
||||
return
|
||||
|
||||
if self._status == "completed":
|
||||
elapsed = self._elapsed_frozen or (time.time() - self._start_time)
|
||||
t.append("✔ ", style=f"bold {_C_GREEN}")
|
||||
@@ -1839,6 +2002,13 @@ FooterKey .footer-key--key {
|
||||
def _subscribe(self) -> None:
|
||||
from crewai.events.event_bus import crewai_event_bus
|
||||
from crewai.events.types.crew_events import CrewKickoffStartedEvent
|
||||
from crewai.events.types.flow_events import (
|
||||
FlowStartedEvent,
|
||||
MethodExecutionFailedEvent,
|
||||
MethodExecutionFinishedEvent,
|
||||
MethodExecutionPausedEvent,
|
||||
MethodExecutionStartedEvent,
|
||||
)
|
||||
from crewai.events.types.llm_events import (
|
||||
LLMCallCompletedEvent,
|
||||
LLMCallStartedEvent,
|
||||
@@ -1872,13 +2042,74 @@ FooterKey .footer-key--key {
|
||||
@crewai_event_bus.on(CrewKickoffStartedEvent)
|
||||
def on_crew_started(source: Any, event: CrewKickoffStartedEvent) -> None:
|
||||
with self._lock:
|
||||
if event.crew_name:
|
||||
# In flow mode the app is named for the flow; a nested crew's
|
||||
# kickoff (a `call: crew` step) must not rename it.
|
||||
if event.crew_name and not self._is_flow_run:
|
||||
self._crew_name = event.crew_name
|
||||
self.title = f"CrewAI — {event.crew_name}"
|
||||
self._status = "working"
|
||||
|
||||
self._register_handler(CrewKickoffStartedEvent, on_crew_started)
|
||||
|
||||
# ── Declarative-flow method events → STEPS panel ────────
|
||||
@crewai_event_bus.on(FlowStartedEvent)
|
||||
def on_flow_started(source: Any, event: FlowStartedEvent) -> None:
|
||||
with self._lock:
|
||||
self._status = "working"
|
||||
|
||||
self._register_handler(FlowStartedEvent, on_flow_started)
|
||||
|
||||
@crewai_event_bus.on(MethodExecutionStartedEvent)
|
||||
def on_method_started(source: Any, event: MethodExecutionStartedEvent) -> None:
|
||||
with self._lock:
|
||||
name = event.method_name
|
||||
self._current_method = name
|
||||
# Agent is per-method; clear it so the header doesn't show the
|
||||
# previous method's agent until a new agent event arrives.
|
||||
self._current_agent = ""
|
||||
for step in self._flow_steps:
|
||||
if step["name"] == name:
|
||||
step["status"] = "active"
|
||||
break
|
||||
else:
|
||||
self._flow_steps.append(
|
||||
{
|
||||
"name": name,
|
||||
"call_type": self._flow_method_types.get(name),
|
||||
"status": "active",
|
||||
}
|
||||
)
|
||||
|
||||
self._register_handler(MethodExecutionStartedEvent, on_method_started)
|
||||
|
||||
@crewai_event_bus.on(MethodExecutionFinishedEvent)
|
||||
def on_method_finished(
|
||||
source: Any, event: MethodExecutionFinishedEvent
|
||||
) -> None:
|
||||
with self._lock:
|
||||
self._set_flow_step_status(event.method_name, "done")
|
||||
self._clear_current_method(event.method_name)
|
||||
|
||||
self._register_handler(MethodExecutionFinishedEvent, on_method_finished)
|
||||
|
||||
@crewai_event_bus.on(MethodExecutionFailedEvent)
|
||||
def on_method_failed(source: Any, event: MethodExecutionFailedEvent) -> None:
|
||||
with self._lock:
|
||||
self._set_flow_step_status(event.method_name, "failed")
|
||||
self._clear_current_method(event.method_name)
|
||||
|
||||
self._register_handler(MethodExecutionFailedEvent, on_method_failed)
|
||||
|
||||
@crewai_event_bus.on(MethodExecutionPausedEvent)
|
||||
def on_method_paused(source: Any, event: MethodExecutionPausedEvent) -> None:
|
||||
# A @human_feedback method paused; flow status panels are suppressed
|
||||
# in TUI mode, so surface the wait in STEPS/header instead of leaving
|
||||
# a spinner. _current_method stays pointed at it.
|
||||
with self._lock:
|
||||
self._set_flow_step_status(event.method_name, "paused")
|
||||
|
||||
self._register_handler(MethodExecutionPausedEvent, on_method_paused)
|
||||
|
||||
@crewai_event_bus.on(TaskStartedEvent)
|
||||
def on_task_started(source: Any, event: TaskStartedEvent) -> None:
|
||||
with self._lock:
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import click
|
||||
from crewai_core.project import ProjectDefinitionError, configured_project_definition
|
||||
@@ -18,6 +19,13 @@ from crewai_cli.input_prompt import (
|
||||
from crewai_cli.utils import build_env_with_all_tool_credentials
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from crewai.flow.flow import Flow
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def run_declarative_flow_in_project_env(
|
||||
definition: str | Path, inputs: str | None = None
|
||||
) -> None:
|
||||
@@ -66,17 +74,182 @@ def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> N
|
||||
flow = load_declarative_flow(definition)
|
||||
resolved_inputs = _resolve_flow_inputs(flow, provided)
|
||||
|
||||
# The TUI is the interactive default. Headless contexts run directly on the
|
||||
# terminal: deploy/CREWAI_DMN, piped output, CI — anything without an
|
||||
# interactive TTY. is_interactive() already folds in the CREWAI_DMN check.
|
||||
# Human-feedback flows also run on the terminal: their methods collect input
|
||||
# via the flow runtime's blocking input()/Rich prompts (and async feedback
|
||||
# returns a pending marker rather than completing), neither of which the
|
||||
# Textual TUI can handle correctly.
|
||||
if is_interactive() and not _flow_uses_human_feedback(flow):
|
||||
_run_declarative_flow_tui(flow, resolved_inputs or None)
|
||||
return
|
||||
|
||||
try:
|
||||
result = flow.kickoff(inputs=resolved_inputs or None)
|
||||
except Exception as exc:
|
||||
click.echo(
|
||||
f"An error occurred while running the declarative flow: {exc}", err=True
|
||||
f"An error occurred while running the declarative flow: {exc}",
|
||||
err=True,
|
||||
)
|
||||
raise SystemExit(1) from exc
|
||||
|
||||
click.echo(_format_result(result))
|
||||
|
||||
|
||||
def _run_declarative_flow_tui(
|
||||
flow: Flow[Any], resolved_inputs: dict[str, Any] | None
|
||||
) -> Any:
|
||||
"""Run a declarative flow on the CrewAI TUI (the interactive default).
|
||||
|
||||
Mirrors the declarative-crew TUI contract (``run_crew._run_json_crew``):
|
||||
a failed flow exits non-zero, a user quit ends the process so in-flight LLM
|
||||
work stops, and choosing Deploy chains into the deploy command.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
from crewai.events.event_listener import EventListener
|
||||
|
||||
from crewai_cli.crew_run_tui import CrewRunApp
|
||||
|
||||
# The flow runtime (unlike a Crew constructor) doesn't create the event
|
||||
# listener, and the TUI's trace/telemetry features depend on it.
|
||||
EventListener()
|
||||
|
||||
# The STEPS panel and header are driven by flow method events. A flow may
|
||||
# declare ``config.suppress_flow_events`` (a headless/production
|
||||
# optimization) which would leave STEPS stuck on "waiting…" here — so force
|
||||
# emission on for the interactive TUI run. The headless path never reaches
|
||||
# this and keeps the flow's declared setting.
|
||||
try:
|
||||
flow.suppress_flow_events = False
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Could not disable suppress_flow_events for the flow TUI", exc_info=True
|
||||
)
|
||||
|
||||
app = CrewRunApp(crew_name=flow.name or type(flow).__name__)
|
||||
app._flow = flow
|
||||
app._flow_inputs = resolved_inputs
|
||||
app._flow_method_types = _flow_method_types(flow)
|
||||
|
||||
app.run()
|
||||
|
||||
_print_flow_post_tui_summary(app)
|
||||
|
||||
if app._status == "failed":
|
||||
raise SystemExit(1)
|
||||
|
||||
if app._status not in ("completed", "failed"):
|
||||
# User quit mid-run. kickoff runs in a thread worker that cannot be
|
||||
# force-cancelled, so end the process to stop in-flight LLM and tool
|
||||
# work instead of letting it burn tokens in the background.
|
||||
click.secho("\n Run cancelled.", fg="yellow")
|
||||
sys.stdout.flush()
|
||||
os._exit(130)
|
||||
|
||||
if getattr(app, "_want_deploy", False):
|
||||
from crewai_cli.run_crew import _chain_deploy
|
||||
|
||||
_chain_deploy()
|
||||
|
||||
return app._crew_result
|
||||
|
||||
|
||||
def _flow_uses_human_feedback(flow: Flow[Any]) -> bool:
|
||||
"""True if any declarative method declares ``@human_feedback``.
|
||||
|
||||
Such flows need the flow runtime's interactive stdin / Rich prompts, which
|
||||
don't compose with Textual — so they run on the terminal, not the TUI.
|
||||
"""
|
||||
try:
|
||||
return any(
|
||||
method.human_feedback is not None
|
||||
for method in flow._definition.methods.values()
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Could not inspect flow for human feedback", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
def _flow_method_types(flow: Flow[Any]) -> dict[str, str]:
|
||||
"""Map each declarative method name to its ``call`` type (crew/agent/…).
|
||||
|
||||
Best-effort: the STEPS panel shows this as a dim label. Method events don't
|
||||
carry the call type, so it's read from the flow definition up front.
|
||||
"""
|
||||
method_types: dict[str, str] = {}
|
||||
try:
|
||||
for name, method_definition in flow._definition.methods.items():
|
||||
method_types[name] = method_definition.do.call
|
||||
except Exception:
|
||||
logger.debug("Could not derive flow method types", exc_info=True)
|
||||
return method_types
|
||||
|
||||
|
||||
def _print_flow_post_tui_summary(app: Any) -> None:
|
||||
"""Print a compact result panel after the flow TUI exits."""
|
||||
import time
|
||||
|
||||
from rich.console import Console
|
||||
from rich.markdown import Markdown
|
||||
from rich.padding import Padding
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
console = Console()
|
||||
elapsed = (app._elapsed_frozen or (time.time() - app._start_time)) or 0.0
|
||||
|
||||
out_tokens = app._output_tokens + app._live_out_tokens
|
||||
token_parts = []
|
||||
if app._input_tokens:
|
||||
token_parts.append(f"↑{app._input_tokens:,}")
|
||||
if out_tokens:
|
||||
token_parts.append(f"↓{out_tokens:,}")
|
||||
token_str = " ".join(token_parts)
|
||||
if token_str:
|
||||
token_str += " tokens"
|
||||
|
||||
crewai_red = "#FF5A50"
|
||||
crewai_teal = "#1F7982"
|
||||
|
||||
if app._status == "completed":
|
||||
summary = Text()
|
||||
summary.append(" ✔ Flow complete", style=f"bold {crewai_teal}")
|
||||
summary.append(f" in {elapsed:.1f}s", style="dim")
|
||||
if token_str:
|
||||
summary.append(f" {token_str}", style="dim")
|
||||
console.print(
|
||||
Panel(
|
||||
summary,
|
||||
title=f" {app._crew_name} ",
|
||||
title_align="left",
|
||||
border_style=crewai_teal,
|
||||
padding=(0, 1),
|
||||
)
|
||||
)
|
||||
if app._final_output:
|
||||
console.print()
|
||||
console.print(Text(" Final Result", style=f"bold {crewai_teal}"))
|
||||
console.print()
|
||||
console.print(Padding(Markdown(app._final_output), (0, 2)))
|
||||
elif app._status == "failed":
|
||||
content = Text()
|
||||
content.append(" ✘ Failed", style=f"bold {crewai_red}")
|
||||
content.append(f" after {elapsed:.1f}s\n", style="dim")
|
||||
if app._error:
|
||||
content.append(f"\n {app._error}\n", style=crewai_red)
|
||||
console.print(
|
||||
Panel(
|
||||
content,
|
||||
title=f" {app._crew_name} ",
|
||||
title_align="left",
|
||||
border_style=crewai_red,
|
||||
padding=(0, 1),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Resolve kickoff inputs from the flow's state schema.
|
||||
|
||||
|
||||
@@ -6,6 +6,14 @@ from unittest.mock import Mock
|
||||
import pytest
|
||||
|
||||
from crewai.events.event_bus import crewai_event_bus
|
||||
from crewai.events.types.crew_events import CrewKickoffStartedEvent
|
||||
from crewai.events.types.flow_events import (
|
||||
FlowStartedEvent,
|
||||
MethodExecutionFailedEvent,
|
||||
MethodExecutionFinishedEvent,
|
||||
MethodExecutionPausedEvent,
|
||||
MethodExecutionStartedEvent,
|
||||
)
|
||||
from crewai.events.types.memory_events import (
|
||||
MemorySaveCompletedEvent,
|
||||
MemorySaveFailedEvent,
|
||||
@@ -959,6 +967,31 @@ async def test_crew_done_does_not_mark_unfinished_tool_successful() -> None:
|
||||
assert app._plan_step_status == {1: "failed", 2: "done", 3: "done"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_flow_done_uses_flow_wording_for_unfinished_tool() -> None:
|
||||
# The shared completion handler reports "flow" (not "crew") in flow mode.
|
||||
app = CrewRunApp(crew_name="Demo Flow")
|
||||
app._flow = SimpleNamespace()
|
||||
|
||||
async with app.run_test(size=(100, 40)) as pilot:
|
||||
app._log_entries = [
|
||||
{
|
||||
"tool_name": "search",
|
||||
"status": "running",
|
||||
"args": None,
|
||||
"result": None,
|
||||
"error": None,
|
||||
"start_time": time.time() - 2,
|
||||
"duration": None,
|
||||
"task_idx": 1,
|
||||
}
|
||||
]
|
||||
app._on_crew_done("final output")
|
||||
await pilot.pause()
|
||||
|
||||
assert app._log_entries[0]["error"] == "No result received before flow completed"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_crew_done_does_not_timeout_memory_save() -> None:
|
||||
app = _app_with_plan()
|
||||
@@ -1481,3 +1514,210 @@ def test_overlapping_task_logs_keep_their_own_state() -> None:
|
||||
assert any(step.get("summary") == "thinking" for step in entry2["steps"])
|
||||
finally:
|
||||
app._unsubscribe()
|
||||
|
||||
|
||||
# ── Declarative-flow (non-conversational) TUI support ───────
|
||||
|
||||
|
||||
def test_is_flow_run_gating() -> None:
|
||||
"""The flow-render gate must be true only for a non-conversational flow."""
|
||||
crew_app = CrewRunApp(total_tasks=1)
|
||||
crew_app._crew = SimpleNamespace()
|
||||
assert crew_app._is_flow_run is False
|
||||
|
||||
conv_app = CrewRunApp(conversational=True)
|
||||
conv_app._flow = SimpleNamespace()
|
||||
assert conv_app._is_flow_run is False
|
||||
|
||||
flow_app = CrewRunApp()
|
||||
flow_app._flow = SimpleNamespace()
|
||||
assert flow_app._is_flow_run is True
|
||||
|
||||
|
||||
def test_flow_method_events_build_steps() -> None:
|
||||
app = CrewRunApp(crew_name="Demo")
|
||||
app._flow = SimpleNamespace()
|
||||
app._flow_method_types = {"research": "crew", "summarize": "agent"}
|
||||
app._subscribe()
|
||||
try:
|
||||
_emit_event(FlowStartedEvent(flow_name="Demo"))
|
||||
assert app._status == "working"
|
||||
|
||||
_emit_event(
|
||||
MethodExecutionStartedEvent(
|
||||
flow_name="Demo", method_name="research", state={}
|
||||
)
|
||||
)
|
||||
assert app._flow_steps == [
|
||||
{"name": "research", "call_type": "crew", "status": "active"}
|
||||
]
|
||||
assert app._current_method == "research"
|
||||
|
||||
_emit_event(
|
||||
MethodExecutionFinishedEvent(
|
||||
flow_name="Demo", method_name="research", result="ok", state={}
|
||||
)
|
||||
)
|
||||
_emit_event(
|
||||
MethodExecutionStartedEvent(
|
||||
flow_name="Demo", method_name="summarize", state={}
|
||||
)
|
||||
)
|
||||
_emit_event(
|
||||
MethodExecutionFailedEvent(
|
||||
flow_name="Demo",
|
||||
method_name="summarize",
|
||||
error=RuntimeError("boom"),
|
||||
)
|
||||
)
|
||||
finally:
|
||||
app._unsubscribe()
|
||||
|
||||
assert app._flow_steps == [
|
||||
{"name": "research", "call_type": "crew", "status": "done"},
|
||||
{"name": "summarize", "call_type": "agent", "status": "failed"},
|
||||
]
|
||||
# The header must not keep spinning a method that already ended.
|
||||
assert app._current_method is None
|
||||
|
||||
|
||||
def test_current_method_clears_and_falls_back_across_overlap() -> None:
|
||||
app = CrewRunApp(crew_name="Demo")
|
||||
app._flow = SimpleNamespace()
|
||||
app._subscribe()
|
||||
try:
|
||||
_emit_event(
|
||||
MethodExecutionStartedEvent(flow_name="Demo", method_name="a", state={})
|
||||
)
|
||||
_emit_event(
|
||||
MethodExecutionStartedEvent(flow_name="Demo", method_name="b", state={})
|
||||
)
|
||||
assert app._current_method == "b"
|
||||
|
||||
# 'a' finishes while 'b' is still active → header stays on 'b'.
|
||||
_emit_event(
|
||||
MethodExecutionFinishedEvent(
|
||||
flow_name="Demo", method_name="a", result=None, state={}
|
||||
)
|
||||
)
|
||||
assert app._current_method == "b"
|
||||
|
||||
# 'b' finishes → nothing active left → header clears.
|
||||
_emit_event(
|
||||
MethodExecutionFinishedEvent(
|
||||
flow_name="Demo", method_name="b", result=None, state={}
|
||||
)
|
||||
)
|
||||
assert app._current_method is None
|
||||
finally:
|
||||
app._unsubscribe()
|
||||
|
||||
|
||||
def test_flow_method_transitions_clear_current_agent() -> None:
|
||||
app = CrewRunApp(crew_name="Demo")
|
||||
app._flow = SimpleNamespace()
|
||||
app._subscribe()
|
||||
try:
|
||||
_emit_event(
|
||||
MethodExecutionStartedEvent(flow_name="Demo", method_name="a", state={})
|
||||
)
|
||||
app._current_agent = "Researcher" # an agent ran during method 'a'
|
||||
|
||||
# Starting a new method clears the previous method's agent.
|
||||
_emit_event(
|
||||
MethodExecutionStartedEvent(flow_name="Demo", method_name="b", state={})
|
||||
)
|
||||
assert app._current_agent == ""
|
||||
|
||||
app._current_agent = "Writer"
|
||||
# 'b' ending switches the active method ('a' still active) → agent clears.
|
||||
_emit_event(
|
||||
MethodExecutionFinishedEvent(
|
||||
flow_name="Demo", method_name="b", result=None, state={}
|
||||
)
|
||||
)
|
||||
assert app._current_agent == ""
|
||||
finally:
|
||||
app._unsubscribe()
|
||||
|
||||
|
||||
def test_crew_kickoff_does_not_rename_flow_run() -> None:
|
||||
# A `call: crew` step must not relabel the flow with the nested crew's name.
|
||||
app = CrewRunApp(crew_name="My Flow")
|
||||
app._flow = SimpleNamespace()
|
||||
app._subscribe()
|
||||
try:
|
||||
_emit_event(CrewKickoffStartedEvent(crew_name="Nested Crew", inputs=None))
|
||||
assert app._crew_name == "My Flow"
|
||||
assert app._status == "working"
|
||||
finally:
|
||||
app._unsubscribe()
|
||||
|
||||
|
||||
def test_crew_kickoff_renames_in_crew_mode() -> None:
|
||||
# Regression: crew runs still adopt the crew name from the event.
|
||||
app = CrewRunApp(crew_name="Crew")
|
||||
app._crew = SimpleNamespace()
|
||||
app._subscribe()
|
||||
try:
|
||||
_emit_event(CrewKickoffStartedEvent(crew_name="Real Crew", inputs=None))
|
||||
assert app._crew_name == "Real Crew"
|
||||
finally:
|
||||
app._unsubscribe()
|
||||
|
||||
|
||||
def test_method_paused_marks_step_paused() -> None:
|
||||
app = CrewRunApp(crew_name="Demo")
|
||||
app._flow = SimpleNamespace()
|
||||
app._subscribe()
|
||||
try:
|
||||
_emit_event(
|
||||
MethodExecutionStartedEvent(flow_name="Demo", method_name="ask", state={})
|
||||
)
|
||||
_emit_event(
|
||||
MethodExecutionPausedEvent(
|
||||
flow_name="Demo",
|
||||
method_name="ask",
|
||||
state={},
|
||||
flow_id="flow-1",
|
||||
message="Need your input",
|
||||
)
|
||||
)
|
||||
assert app._flow_steps == [
|
||||
{"name": "ask", "call_type": None, "status": "paused"}
|
||||
]
|
||||
assert app._current_method == "ask"
|
||||
finally:
|
||||
app._unsubscribe()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_declarative_flow_runs_on_tui() -> None:
|
||||
"""End-to-end: on_mount dispatches _run_flow_worker → flow.kickoff →
|
||||
_on_crew_done, and any still-active step is swept to done on completion."""
|
||||
kicked: dict[str, object] = {}
|
||||
|
||||
class FakeFlow:
|
||||
name = "Demo Flow"
|
||||
|
||||
def kickoff(self, inputs=None):
|
||||
kicked["inputs"] = inputs
|
||||
return "flow result"
|
||||
|
||||
app = CrewRunApp(crew_name="Demo Flow")
|
||||
app._flow = FakeFlow()
|
||||
app._flow_inputs = {"topic": "AI"}
|
||||
# A step left active (no Finished event) must be swept to done by _on_crew_done.
|
||||
app._flow_steps = [{"name": "compute", "call_type": "expression", "status": "active"}]
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
for _ in range(100):
|
||||
await pilot.pause(0.05)
|
||||
if app._status == "completed":
|
||||
break
|
||||
|
||||
assert kicked["inputs"] == {"topic": "AI"}
|
||||
assert app._status == "completed"
|
||||
assert app._final_output == "flow result"
|
||||
assert app._crew_result == "flow result"
|
||||
assert app._flow_steps[0]["status"] == "done"
|
||||
|
||||
@@ -2,6 +2,7 @@ from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -9,6 +10,17 @@ import crewai_cli.input_prompt as input_prompt_module
|
||||
import crewai_cli.run_declarative_flow as run_declarative_flow_module
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _headless_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Default these tests to the headless/terminal path.
|
||||
|
||||
``run_declarative_flow`` now launches the TUI when interactive, which can't
|
||||
run under pytest; tests here assert the terminal/headless contract. Tests
|
||||
that exercise TUI routing override ``is_dmn_mode_enabled`` explicitly.
|
||||
"""
|
||||
monkeypatch.setenv("CREWAI_DMN", "true")
|
||||
|
||||
|
||||
FLOW_YAML = """\
|
||||
schema: crewai.flow/v1
|
||||
name: TestFlow
|
||||
@@ -400,3 +412,202 @@ def test_id_restore_still_drops_unknown_keys(
|
||||
assert resolved == {"id": "run-123"} # id kept, typo dropped
|
||||
assert "Ignoring unknown input 'prospect_emai'" in captured.err
|
||||
assert "Ignoring unknown input 'id'" not in captured.err
|
||||
|
||||
|
||||
# ── TUI vs terminal (headless/deploy) routing ──────────────────────
|
||||
|
||||
|
||||
def _install_fake_flow_app(monkeypatch, *, status, want_deploy=False):
|
||||
"""Replace CrewRunApp/EventListener/summary so _run_declarative_flow_tui is
|
||||
driven by a controllable fake app."""
|
||||
|
||||
class FakeEventListener:
|
||||
pass
|
||||
|
||||
class FakeApp:
|
||||
def __init__(self, crew_name=""):
|
||||
self._crew_name = crew_name
|
||||
self._status = status
|
||||
self._want_deploy = want_deploy
|
||||
self._crew_result = "result"
|
||||
|
||||
def run(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(
|
||||
"crewai.events.event_listener.EventListener", FakeEventListener
|
||||
)
|
||||
monkeypatch.setattr("crewai_cli.crew_run_tui.CrewRunApp", FakeApp)
|
||||
monkeypatch.setattr(
|
||||
run_declarative_flow_module, "_print_flow_post_tui_summary", lambda app: None
|
||||
)
|
||||
|
||||
|
||||
def test_run_declarative_flow_dmn_uses_terminal(
|
||||
tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("CREWAI_DMN", "true")
|
||||
monkeypatch.setattr(
|
||||
run_declarative_flow_module,
|
||||
"_run_declarative_flow_tui",
|
||||
lambda *a, **k: pytest.fail("DMN/headless mode must not launch the TUI"),
|
||||
)
|
||||
path = _write(tmp_path, REQUIRED_FLOW_YAML)
|
||||
|
||||
run_declarative_flow_module.run_declarative_flow(
|
||||
str(path), '{"prospect_email":"a@b.com"}'
|
||||
)
|
||||
|
||||
assert capsys.readouterr().out == "a@b.com\n"
|
||||
|
||||
|
||||
def test_run_declarative_flow_interactive_uses_tui(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setattr(run_declarative_flow_module, "is_interactive", lambda: True)
|
||||
captured: dict[str, object] = {}
|
||||
monkeypatch.setattr(
|
||||
run_declarative_flow_module,
|
||||
"_run_declarative_flow_tui",
|
||||
lambda flow, resolved: captured.update(flow=flow, inputs=resolved),
|
||||
)
|
||||
path = _write(tmp_path, REQUIRED_FLOW_YAML)
|
||||
|
||||
run_declarative_flow_module.run_declarative_flow(
|
||||
str(path), '{"prospect_email":"a@b.com"}'
|
||||
)
|
||||
|
||||
assert captured["inputs"] == {"prospect_email": "a@b.com"}
|
||||
assert captured["flow"] is not None
|
||||
|
||||
|
||||
def test_run_declarative_flow_tui_failed_exits_nonzero(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_install_fake_flow_app(monkeypatch, status="failed")
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
run_declarative_flow_module._run_declarative_flow_tui(
|
||||
SimpleNamespace(name="Flow"), None
|
||||
)
|
||||
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
def test_run_declarative_flow_tui_user_quit_exits_130(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_install_fake_flow_app(monkeypatch, status="chatting")
|
||||
exit_calls: list[int] = []
|
||||
monkeypatch.setattr(os, "_exit", lambda code: exit_calls.append(code))
|
||||
|
||||
run_declarative_flow_module._run_declarative_flow_tui(
|
||||
SimpleNamespace(name="Flow"), None
|
||||
)
|
||||
|
||||
assert exit_calls == [130]
|
||||
|
||||
|
||||
def test_run_declarative_flow_tui_chains_deploy(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_install_fake_flow_app(monkeypatch, status="completed", want_deploy=True)
|
||||
deploy_calls: list[bool] = []
|
||||
monkeypatch.setattr(
|
||||
"crewai_cli.run_crew._chain_deploy", lambda: deploy_calls.append(True)
|
||||
)
|
||||
|
||||
run_declarative_flow_module._run_declarative_flow_tui(
|
||||
SimpleNamespace(name="Flow"), None
|
||||
)
|
||||
|
||||
assert deploy_calls == [True]
|
||||
|
||||
|
||||
def test_run_declarative_flow_tui_no_deploy_when_not_requested(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_install_fake_flow_app(monkeypatch, status="completed", want_deploy=False)
|
||||
deploy_calls: list[bool] = []
|
||||
monkeypatch.setattr(
|
||||
"crewai_cli.run_crew._chain_deploy", lambda: deploy_calls.append(True)
|
||||
)
|
||||
|
||||
run_declarative_flow_module._run_declarative_flow_tui(
|
||||
SimpleNamespace(name="Flow"), None
|
||||
)
|
||||
|
||||
assert deploy_calls == []
|
||||
|
||||
|
||||
def test_run_declarative_flow_tui_enables_flow_events(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
# The STEPS panel depends on flow method events; a flow that declared
|
||||
# suppress_flow_events must have it forced off for the interactive TUI run.
|
||||
_install_fake_flow_app(monkeypatch, status="completed")
|
||||
flow = SimpleNamespace(name="Flow", suppress_flow_events=True)
|
||||
|
||||
run_declarative_flow_module._run_declarative_flow_tui(flow, None)
|
||||
|
||||
assert flow.suppress_flow_events is False
|
||||
|
||||
|
||||
def test_flow_uses_human_feedback_detection() -> None:
|
||||
hf_flow = SimpleNamespace(
|
||||
_definition=SimpleNamespace(
|
||||
methods={
|
||||
"ask": SimpleNamespace(human_feedback=SimpleNamespace(emit=None)),
|
||||
"plain": SimpleNamespace(human_feedback=None),
|
||||
}
|
||||
)
|
||||
)
|
||||
assert run_declarative_flow_module._flow_uses_human_feedback(hf_flow) is True
|
||||
|
||||
no_hf = SimpleNamespace(
|
||||
_definition=SimpleNamespace(
|
||||
methods={"a": SimpleNamespace(human_feedback=None)}
|
||||
)
|
||||
)
|
||||
assert run_declarative_flow_module._flow_uses_human_feedback(no_hf) is False
|
||||
# No definition → False, no error.
|
||||
assert run_declarative_flow_module._flow_uses_human_feedback(SimpleNamespace()) is False
|
||||
|
||||
|
||||
def test_human_feedback_flow_uses_terminal_even_when_interactive(
|
||||
tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
# A human-feedback flow must run on the terminal (blocking input / Rich
|
||||
# prompts) even in an interactive session, never on the TUI.
|
||||
monkeypatch.setattr(run_declarative_flow_module, "is_interactive", lambda: True)
|
||||
monkeypatch.setattr(
|
||||
run_declarative_flow_module, "_flow_uses_human_feedback", lambda flow: True
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
run_declarative_flow_module,
|
||||
"_run_declarative_flow_tui",
|
||||
lambda *a, **k: pytest.fail("human-feedback flow must run on the terminal"),
|
||||
)
|
||||
path = _write(tmp_path, FLOW_YAML)
|
||||
|
||||
run_declarative_flow_module.run_declarative_flow(str(path), '{"topic":"AI"}')
|
||||
|
||||
assert capsys.readouterr().out == "AI\n"
|
||||
|
||||
|
||||
def test_flow_method_types_from_definition() -> None:
|
||||
flow = SimpleNamespace(
|
||||
_definition=SimpleNamespace(
|
||||
methods={
|
||||
"fetch": SimpleNamespace(do=SimpleNamespace(call="expression")),
|
||||
"research": SimpleNamespace(do=SimpleNamespace(call="crew")),
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
assert run_declarative_flow_module._flow_method_types(flow) == {
|
||||
"fetch": "expression",
|
||||
"research": "crew",
|
||||
}
|
||||
# No definition → empty map, no error.
|
||||
assert run_declarative_flow_module._flow_method_types(SimpleNamespace()) == {}
|
||||
|
||||
@@ -22,7 +22,7 @@ dependencies = [
|
||||
"opentelemetry-sdk~=1.42.0",
|
||||
"opentelemetry-exporter-otlp-proto-http~=1.42.0",
|
||||
# Data Handling
|
||||
"chromadb~=1.1.0",
|
||||
"chromadb~=1.5.9",
|
||||
"tokenizers>=0.21,<1",
|
||||
"openpyxl~=3.1.5",
|
||||
# Authentication and Security
|
||||
|
||||
@@ -86,6 +86,7 @@ from crewai.skills.models import Skill as SkillModel
|
||||
from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
|
||||
from crewai.tools.agent_tools.agent_tools import AgentTools
|
||||
from crewai.types.callback import SerializableCallable
|
||||
from crewai.types.usage_metrics import UsageMetrics
|
||||
from crewai.utilities.agent_utils import (
|
||||
get_tool_names,
|
||||
is_inside_event_loop,
|
||||
@@ -400,10 +401,29 @@ class Agent(BaseAgent):
|
||||
return self.planning_config is not None or self.planning
|
||||
|
||||
def _setup_agent_executor(self) -> None:
|
||||
"""Initialize the agent executor with a default cache handler."""
|
||||
if not self.cache_handler:
|
||||
self.cache_handler = CacheHandler()
|
||||
self.set_cache_handler(self.cache_handler)
|
||||
"""Initialize the agent's tools handler and optional tool cache.
|
||||
|
||||
Tool-result caching is opt-in: a standalone agent gets a cache only
|
||||
when it was constructed with an explicit ``cache=True`` or a
|
||||
``cache_handler``. Agents inside a crew additionally receive the
|
||||
crew's shared handler when ``Crew(cache=True)``. Without an opt-in,
|
||||
repeated tool calls with identical arguments always re-execute the
|
||||
tool — the safe default for live-data and state-mutating tools.
|
||||
"""
|
||||
# Recorded before any crew can offer its shared handler at kickoff,
|
||||
# so copy() can distinguish a construction-time opt-in from runtime
|
||||
# crew wiring (which must not turn copies into cachers).
|
||||
self._constructor_cache_opt_in = bool(
|
||||
self.cache
|
||||
and (self.cache_handler is not None or "cache" in self.model_fields_set)
|
||||
)
|
||||
opted_in = self.cache_handler is not None or (
|
||||
"cache" in self.model_fields_set and self.cache
|
||||
)
|
||||
if opted_in:
|
||||
if not self.cache_handler:
|
||||
self.cache_handler = CacheHandler()
|
||||
self.set_cache_handler(self.cache_handler)
|
||||
|
||||
def set_knowledge(self, crew_embedder: EmbedderConfig | None = None) -> None:
|
||||
"""Initialize knowledge sources with the agent or crew embedder config."""
|
||||
@@ -1582,9 +1602,18 @@ class Agent(BaseAgent):
|
||||
crewai_event_bus.emit(self, event=started_event)
|
||||
self._kickoff_event_id = started_event.event_id
|
||||
|
||||
output = self._execute_and_build_output(executor, inputs, response_format)
|
||||
usage_baseline = self._current_usage_summary()
|
||||
output = self._execute_and_build_output(
|
||||
executor, inputs, response_format, usage_baseline
|
||||
)
|
||||
return self._finalize_kickoff(
|
||||
output, executor, inputs, response_format, messages, agent_info
|
||||
output,
|
||||
executor,
|
||||
inputs,
|
||||
response_format,
|
||||
messages,
|
||||
agent_info,
|
||||
usage_baseline,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
@@ -1598,6 +1627,7 @@ class Agent(BaseAgent):
|
||||
response_format: type[Any] | None,
|
||||
messages: str | list[LLMMessage],
|
||||
agent_info: dict[str, Any],
|
||||
usage_baseline: UsageMetrics | None = None,
|
||||
) -> LiteAgentOutput:
|
||||
"""Apply guardrails, save to memory, and emit completion event.
|
||||
|
||||
@@ -1608,6 +1638,8 @@ class Agent(BaseAgent):
|
||||
response_format: Optional response format.
|
||||
messages: The original messages.
|
||||
agent_info: Agent metadata for events.
|
||||
usage_baseline: Usage snapshot taken at kickoff start, so retries
|
||||
report per-call usage relative to it.
|
||||
|
||||
Returns:
|
||||
The finalized output.
|
||||
@@ -1618,6 +1650,7 @@ class Agent(BaseAgent):
|
||||
executor=executor,
|
||||
inputs=inputs,
|
||||
response_format=response_format,
|
||||
usage_baseline=usage_baseline,
|
||||
)
|
||||
|
||||
self._save_kickoff_to_memory(messages, output.raw)
|
||||
@@ -1669,11 +1702,24 @@ class Agent(BaseAgent):
|
||||
except Exception as e:
|
||||
self._logger.log("error", f"Failed to save kickoff result to memory: {e}")
|
||||
|
||||
def _current_usage_summary(self) -> UsageMetrics:
|
||||
"""Snapshot the cumulative usage counters backing this agent's LLM.
|
||||
|
||||
The counters live on the LLM instance (or the agent's token process
|
||||
for non-BaseLLM models) and grow for the object's lifetime — across
|
||||
calls and across agents sharing the instance. Per-call usage is the
|
||||
delta between two snapshots.
|
||||
"""
|
||||
if isinstance(self.llm, BaseLLM):
|
||||
return self.llm.get_token_usage_summary()
|
||||
return self._token_process.get_summary()
|
||||
|
||||
def _build_output_from_result(
|
||||
self,
|
||||
result: dict[str, Any],
|
||||
executor: AgentExecutor,
|
||||
response_format: type[Any] | None = None,
|
||||
usage_baseline: UsageMetrics | None = None,
|
||||
) -> LiteAgentOutput:
|
||||
"""Build a LiteAgentOutput from an executor result dict.
|
||||
|
||||
@@ -1683,6 +1729,9 @@ class Agent(BaseAgent):
|
||||
result: The result dictionary from executor.invoke / invoke_async.
|
||||
executor: The executor instance.
|
||||
response_format: Optional response format.
|
||||
usage_baseline: Usage snapshot taken at kickoff start. When given,
|
||||
the output carries only this call's usage (the delta) instead
|
||||
of the LLM instance's cumulative lifetime counters.
|
||||
|
||||
Returns:
|
||||
LiteAgentOutput with raw output, formatted result, and metrics.
|
||||
@@ -1727,10 +1776,9 @@ class Agent(BaseAgent):
|
||||
else:
|
||||
raw_output = str(output) if not isinstance(output, str) else output
|
||||
|
||||
if isinstance(self.llm, BaseLLM):
|
||||
usage_metrics = self.llm.get_token_usage_summary()
|
||||
else:
|
||||
usage_metrics = self._token_process.get_summary()
|
||||
usage_metrics = self._current_usage_summary()
|
||||
if usage_baseline is not None:
|
||||
usage_metrics = usage_metrics.delta_since(usage_baseline)
|
||||
|
||||
raw_str = (
|
||||
raw_output
|
||||
@@ -1759,20 +1807,26 @@ class Agent(BaseAgent):
|
||||
executor: AgentExecutor,
|
||||
inputs: dict[str, str],
|
||||
response_format: type[Any] | None = None,
|
||||
usage_baseline: UsageMetrics | None = None,
|
||||
) -> LiteAgentOutput:
|
||||
"""Execute the agent synchronously and build the output object."""
|
||||
result = cast(dict[str, Any], executor.invoke(inputs))
|
||||
return self._build_output_from_result(result, executor, response_format)
|
||||
return self._build_output_from_result(
|
||||
result, executor, response_format, usage_baseline
|
||||
)
|
||||
|
||||
async def _execute_and_build_output_async(
|
||||
self,
|
||||
executor: AgentExecutor,
|
||||
inputs: dict[str, str],
|
||||
response_format: type[Any] | None = None,
|
||||
usage_baseline: UsageMetrics | None = None,
|
||||
) -> LiteAgentOutput:
|
||||
"""Execute the agent asynchronously and build the output object."""
|
||||
result = await executor.invoke_async(inputs)
|
||||
return self._build_output_from_result(result, executor, response_format)
|
||||
return self._build_output_from_result(
|
||||
result, executor, response_format, usage_baseline
|
||||
)
|
||||
|
||||
def _process_kickoff_guardrail(
|
||||
self,
|
||||
@@ -1781,6 +1835,7 @@ class Agent(BaseAgent):
|
||||
inputs: dict[str, str],
|
||||
response_format: type[Any] | None = None,
|
||||
retry_count: int = 0,
|
||||
usage_baseline: UsageMetrics | None = None,
|
||||
) -> LiteAgentOutput:
|
||||
"""Process guardrail for kickoff execution with retry logic.
|
||||
|
||||
@@ -1790,6 +1845,9 @@ class Agent(BaseAgent):
|
||||
inputs: Input dictionary for re-execution.
|
||||
response_format: Optional response format.
|
||||
retry_count: Current retry count.
|
||||
usage_baseline: Usage snapshot taken at kickoff start, so a
|
||||
retried output reports the whole call's usage, not just the
|
||||
last attempt's.
|
||||
|
||||
Returns:
|
||||
Validated/updated output.
|
||||
@@ -1827,7 +1885,9 @@ class Agent(BaseAgent):
|
||||
role="user",
|
||||
)
|
||||
|
||||
output = self._execute_and_build_output(executor, inputs, response_format)
|
||||
output = self._execute_and_build_output(
|
||||
executor, inputs, response_format, usage_baseline
|
||||
)
|
||||
|
||||
return self._process_kickoff_guardrail(
|
||||
output=output,
|
||||
@@ -1835,6 +1895,7 @@ class Agent(BaseAgent):
|
||||
inputs=inputs,
|
||||
response_format=response_format,
|
||||
retry_count=retry_count + 1,
|
||||
usage_baseline=usage_baseline,
|
||||
)
|
||||
|
||||
if guardrail_result.result is not None:
|
||||
@@ -1897,11 +1958,18 @@ class Agent(BaseAgent):
|
||||
crewai_event_bus.emit(self, event=started_event)
|
||||
self._kickoff_event_id = started_event.event_id
|
||||
|
||||
usage_baseline = self._current_usage_summary()
|
||||
output = await self._execute_and_build_output_async(
|
||||
executor, inputs, response_format
|
||||
executor, inputs, response_format, usage_baseline
|
||||
)
|
||||
return self._finalize_kickoff(
|
||||
output, executor, inputs, response_format, messages, agent_info
|
||||
output,
|
||||
executor,
|
||||
inputs,
|
||||
response_format,
|
||||
messages,
|
||||
agent_info,
|
||||
usage_baseline,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
|
||||
@@ -205,7 +205,11 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
|
||||
role (str): Role of the agent.
|
||||
goal (str): Objective of the agent.
|
||||
backstory (str): Backstory of the agent.
|
||||
cache (bool): Whether the agent should use a cache for tool usage.
|
||||
cache (bool): Whether the agent participates in tool-result caching
|
||||
when a cache is enabled. The default (True) only permits
|
||||
participation — caching activates when the crew sets cache=True
|
||||
or the agent explicitly opts in with cache=True or a
|
||||
cache_handler; cache=False excludes the agent entirely.
|
||||
config (dict[str, Any] | None): Configuration for the agent.
|
||||
verbose (bool): Verbose mode for the Agent Execution.
|
||||
max_rpm (int | None): Maximum number of requests per minute for the agent execution.
|
||||
@@ -254,6 +258,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
|
||||
_logger: Logger = PrivateAttr(default_factory=lambda: Logger(verbose=False))
|
||||
_rpm_controller: RPMController | None = PrivateAttr(default=None)
|
||||
_request_within_rpm_limit: SerializableCallable | None = PrivateAttr(default=None)
|
||||
_constructor_cache_opt_in: bool = PrivateAttr(default=False)
|
||||
_original_role: str | None = PrivateAttr(default=None)
|
||||
_original_goal: str | None = PrivateAttr(default=None)
|
||||
_original_backstory: str | None = PrivateAttr(default=None)
|
||||
@@ -267,7 +272,14 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
|
||||
description="Configuration for the agent", default=None, exclude=True
|
||||
)
|
||||
cache: bool = Field(
|
||||
default=True, description="Whether the agent should use a cache for tool usage."
|
||||
default=True,
|
||||
description=(
|
||||
"Whether the agent participates in tool-result caching when a "
|
||||
"cache is enabled. Caching itself is opt-in: it activates only "
|
||||
"when the crew sets cache=True or the agent explicitly opts in "
|
||||
"(cache=True or a cache_handler at construction). Set False to "
|
||||
"exclude this agent even when the crew enables caching."
|
||||
),
|
||||
)
|
||||
verbose: bool = Field(
|
||||
default=False, description="Verbose mode for the Agent Execution"
|
||||
@@ -716,6 +728,19 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
|
||||
|
||||
copied_data = self.model_dump(exclude=exclude)
|
||||
copied_data = {k: v for k, v in copied_data.items() if v is not None}
|
||||
# Tool-result caching distinguishes "explicitly enabled" from the
|
||||
# field default via model_fields_set; don't let the dump turn the
|
||||
# default into an explicit opt-in on the copy. An agent that opted
|
||||
# in at construction via an explicit cache_handler (excluded from
|
||||
# the dump) must stay opted in — carry the consent as cache=True so
|
||||
# the copy wires its own fresh handler. A handler merely offered by
|
||||
# a crew at kickoff is runtime wiring, not consent, and must not
|
||||
# opt the copy in; _constructor_cache_opt_in is recorded before any
|
||||
# crew wiring can happen.
|
||||
if "cache" not in self.model_fields_set:
|
||||
copied_data.pop("cache", None)
|
||||
if self._constructor_cache_opt_in:
|
||||
copied_data["cache"] = True
|
||||
return type(self)(
|
||||
**copied_data,
|
||||
llm=existing_llm,
|
||||
|
||||
@@ -168,8 +168,11 @@ class Crew(FlowTrackable, BaseModel):
|
||||
manager_agent: Custom agent that will be used as manager.
|
||||
memory: Whether the crew should use memory to store memories of it's
|
||||
execution.
|
||||
cache: Whether the crew should use a cache to store the results of the
|
||||
tools execution.
|
||||
cache: Whether to cache tool results for the crew's agents. Off by
|
||||
default; when enabled, repeated calls to the same tool with
|
||||
identical arguments reuse the first result without re-executing —
|
||||
avoid enabling for live-data or state-mutating tools unless they
|
||||
gate writes with a cache_function.
|
||||
function_calling_llm: The language model that will run the tool calling
|
||||
for all the agents.
|
||||
process: The process flow that the crew will follow (e.g., sequential,
|
||||
@@ -216,7 +219,16 @@ class Crew(FlowTrackable, BaseModel):
|
||||
_kickoff_event_id: str | None = PrivateAttr(default=None)
|
||||
|
||||
name: str | None = Field(default="crew")
|
||||
cache: bool = Field(default=True)
|
||||
cache: bool = Field(
|
||||
default=False,
|
||||
description=(
|
||||
"Whether to cache tool results for the crew's agents. Opt-in: "
|
||||
"when enabled, repeated calls to the same tool with identical "
|
||||
"arguments return the first result without re-executing the "
|
||||
"tool — do not enable for live-data or state-mutating tools "
|
||||
"unless they set a cache_function that prevents caching."
|
||||
),
|
||||
)
|
||||
tasks: list[Task] = Field(default_factory=list)
|
||||
agents: Annotated[
|
||||
list[BaseAgent],
|
||||
@@ -1048,8 +1060,9 @@ class Crew(FlowTrackable, BaseModel):
|
||||
)
|
||||
raise
|
||||
finally:
|
||||
if self._memory is not None and hasattr(self._memory, "drain_writes"):
|
||||
self._memory.drain_writes()
|
||||
# Safety net for the exception path; the success path already
|
||||
# drained in _create_crew_output before emitting completion.
|
||||
self._drain_memory_writes()
|
||||
clear_files(self.id)
|
||||
detach(token)
|
||||
crewai_event_bus._exit_runtime_scope(runtime_scope)
|
||||
@@ -1260,6 +1273,9 @@ class Crew(FlowTrackable, BaseModel):
|
||||
)
|
||||
raise
|
||||
finally:
|
||||
# Safety net for the exception path; the success path already
|
||||
# drained in _create_crew_output before emitting completion.
|
||||
self._drain_memory_writes()
|
||||
clear_files(self.id)
|
||||
detach(token)
|
||||
crewai_event_bus._exit_runtime_scope(runtime_scope)
|
||||
@@ -1503,6 +1519,11 @@ class Crew(FlowTrackable, BaseModel):
|
||||
)
|
||||
self.manager_agent = manager
|
||||
manager.crew = self
|
||||
# The manager is created outside the agents loop that offers the
|
||||
# crew's cache handler at validation time; offer it here so an
|
||||
# opted-in crew (cache=True) also dedupes the manager's tool calls.
|
||||
if self.cache:
|
||||
manager.set_cache_handler(self._cache_handler)
|
||||
|
||||
def _get_execution_start_index(self, tasks: list[Task]) -> int | None:
|
||||
if self.checkpoint_kickoff_event_id is None:
|
||||
@@ -1841,6 +1862,38 @@ class Crew(FlowTrackable, BaseModel):
|
||||
output=output.raw,
|
||||
)
|
||||
|
||||
def _drain_memory_writes(self) -> None:
|
||||
"""Block until all pending background memory saves have completed.
|
||||
|
||||
Covers the crew memory, per-agent memories, and the manager agent's
|
||||
memory — agents save through ``agent.memory`` when set (see
|
||||
``BaseAgentExecutor._save_to_memory``), so draining only
|
||||
``self._memory`` can miss in-flight saves. Scope/slice views are
|
||||
unwrapped to their backing ``Memory`` so each pool is drained once.
|
||||
|
||||
Must run before ``CrewKickoffCompletedEvent`` is emitted: listeners
|
||||
(e.g. telemetry sessions) tear down on that event, and any
|
||||
``MemorySaveCompletedEvent``/``MemorySaveFailedEvent`` emitted after
|
||||
teardown is lost, leaving the save span orphaned.
|
||||
"""
|
||||
seen: set[int] = set()
|
||||
candidates = [
|
||||
self._memory,
|
||||
self.memory,
|
||||
getattr(self.manager_agent, "memory", None),
|
||||
*(getattr(agent, "memory", None) for agent in self.agents),
|
||||
]
|
||||
for mem in candidates:
|
||||
if mem is None or isinstance(mem, bool):
|
||||
continue
|
||||
backing = getattr(mem, "_memory", None) or mem
|
||||
if id(backing) in seen:
|
||||
continue
|
||||
seen.add(id(backing))
|
||||
drain = getattr(backing, "drain_writes", None)
|
||||
if callable(drain):
|
||||
drain()
|
||||
|
||||
def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput:
|
||||
if not task_outputs:
|
||||
raise ValueError("No task outputs available to create crew output.")
|
||||
@@ -1853,6 +1906,10 @@ class Crew(FlowTrackable, BaseModel):
|
||||
final_string_output = final_task_output.raw
|
||||
self._finish_execution(final_string_output)
|
||||
self.token_usage = self.calculate_usage_metrics()
|
||||
# Ensure background memory saves finish (and emit their
|
||||
# completed/failed events) before the kickoff-completed event below
|
||||
# triggers listener teardown/finalization.
|
||||
self._drain_memory_writes()
|
||||
crewai_event_bus.flush()
|
||||
crewai_event_bus.emit(
|
||||
self,
|
||||
|
||||
@@ -24,9 +24,23 @@ class CrewOutput(BaseModel):
|
||||
description="Output of each task", default_factory=list
|
||||
)
|
||||
token_usage: UsageMetrics = Field(
|
||||
description="Processed token summary", default_factory=UsageMetrics
|
||||
description=(
|
||||
"Processed token summary; ``usage_metrics`` exposes the same "
|
||||
"data as a plain dict"
|
||||
),
|
||||
default_factory=UsageMetrics,
|
||||
)
|
||||
|
||||
@property
|
||||
def usage_metrics(self) -> dict[str, Any]:
|
||||
"""Token usage as a plain dict.
|
||||
|
||||
Same attribute name and shape as ``LiteAgentOutput.usage_metrics``
|
||||
(the ``Agent.kickoff()`` result), so a usage accessor written for one
|
||||
result type works on both.
|
||||
"""
|
||||
return self.token_usage.model_dump()
|
||||
|
||||
@property
|
||||
def json(self) -> str | None: # type: ignore[override]
|
||||
if self.tasks_output[-1].output_format != OutputFormat.JSON:
|
||||
|
||||
@@ -211,6 +211,13 @@ To enable tracing, do any one of these:
|
||||
"""Print a panel with consistent formatting if verbose is enabled."""
|
||||
panel = self.create_panel(content, title, style)
|
||||
if is_flow:
|
||||
# A TUI (e.g. the CLI's CrewRunApp) owns the screen and renders flow
|
||||
# progress in its own STEPS panel; emitting Rich panels here would
|
||||
# interleave with and corrupt the TUI, so suppress them in TUI mode.
|
||||
from crewai.events.listeners.tracing.utils import is_tui_mode
|
||||
|
||||
if is_tui_mode():
|
||||
return
|
||||
self.print(panel)
|
||||
self.print()
|
||||
else:
|
||||
|
||||
@@ -133,6 +133,8 @@ class _ConversationalMixin:
|
||||
_pending_user_message: str | dict[str, Any] | None
|
||||
_pending_intents: Sequence[str] | None
|
||||
_pending_intent_llm: str | BaseLLM | None
|
||||
_turn_classified_intent: str | None
|
||||
_assistant_reply_appended: bool
|
||||
|
||||
def _clear_or_listeners(self) -> None:
|
||||
pass
|
||||
@@ -185,12 +187,22 @@ class _ConversationalMixin:
|
||||
)
|
||||
return configured_route
|
||||
|
||||
if state.last_intent:
|
||||
turn_intent = self._turn_classified_intent
|
||||
if turn_intent:
|
||||
state.last_intent = turn_intent
|
||||
self._emit_conversation_route_selected(
|
||||
state.last_intent,
|
||||
turn_intent,
|
||||
previous_intent=previous_intent,
|
||||
)
|
||||
return state.last_intent
|
||||
return turn_intent
|
||||
|
||||
if previous_intent:
|
||||
logger.debug(
|
||||
"route_turn() returned no route and no intent was classified "
|
||||
"this turn; ignoring stale last_intent=%r from a previous turn "
|
||||
"and falling back to built-in routing",
|
||||
previous_intent,
|
||||
)
|
||||
|
||||
if self.can_answer_from_history(context):
|
||||
state.last_intent = "answer_from_history"
|
||||
@@ -310,11 +322,11 @@ class _ConversationalMixin:
|
||||
if "from_checkpoint" not in kickoff_kwargs:
|
||||
self._reset_turn_execution_state()
|
||||
|
||||
assistant_count = self._assistant_message_count()
|
||||
object.__setattr__(self, "_assistant_reply_appended", False)
|
||||
result = self.kickoff(inputs={"id": sid}, **kickoff_kwargs)
|
||||
if (
|
||||
result is not None
|
||||
and self._assistant_message_count() == assistant_count
|
||||
and not self._assistant_reply_appended
|
||||
and self._is_public_turn_result(result)
|
||||
):
|
||||
self.append_assistant_message(self._stringify_result(result))
|
||||
@@ -387,7 +399,7 @@ class _ConversationalMixin:
|
||||
if "from_checkpoint" not in kickoff_kwargs:
|
||||
self._reset_turn_execution_state()
|
||||
|
||||
assistant_count = self._assistant_message_count()
|
||||
object.__setattr__(self, "_assistant_reply_appended", False)
|
||||
original_stream = bool(getattr(self, "stream", False))
|
||||
original_streaming_turn = getattr(
|
||||
self, "_streaming_conversation_turn", False
|
||||
@@ -403,7 +415,7 @@ class _ConversationalMixin:
|
||||
)
|
||||
if (
|
||||
result is not None
|
||||
and self._assistant_message_count() == assistant_count
|
||||
and not self._assistant_reply_appended
|
||||
and self._is_public_turn_result(result)
|
||||
):
|
||||
self.append_assistant_message(self._stringify_result(result))
|
||||
@@ -550,6 +562,11 @@ class _ConversationalMixin:
|
||||
supply per-route descriptions, or change the default/fallback intent.
|
||||
Override this method to bypass the LLM router entirely (e.g.,
|
||||
permission gates before the LLM decision).
|
||||
|
||||
Returning a falsy value means "no routing decision": the turn falls
|
||||
through to the built-in defaults (``answer_from_history`` when
|
||||
configured, else ``converse``). It never replays a previous turn's
|
||||
intent.
|
||||
"""
|
||||
config = self._conversation_config
|
||||
if config is None:
|
||||
@@ -618,6 +635,9 @@ class _ConversationalMixin:
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Append a final user-visible assistant message."""
|
||||
# Explicit signal for handle_turn's "did the handler reply?" check.
|
||||
# A count heuristic breaks when handlers trim history mid-turn.
|
||||
object.__setattr__(self, "_assistant_reply_appended", True)
|
||||
state = cast(ConversationState, self.state)
|
||||
state.messages.append(
|
||||
ConversationMessage(
|
||||
@@ -722,6 +742,7 @@ class _ConversationalMixin:
|
||||
context=self.conversation_messages,
|
||||
)
|
||||
state.last_intent = intent
|
||||
object.__setattr__(self, "_turn_classified_intent", intent)
|
||||
return intent
|
||||
return text
|
||||
|
||||
@@ -788,6 +809,10 @@ class _ConversationalMixin:
|
||||
object.__setattr__(self, "_pending_intent_llm", None)
|
||||
if not hasattr(self, "_streaming_conversation_turn"):
|
||||
object.__setattr__(self, "_streaming_conversation_turn", False)
|
||||
if not hasattr(self, "_turn_classified_intent"):
|
||||
object.__setattr__(self, "_turn_classified_intent", None)
|
||||
if not hasattr(self, "_assistant_reply_appended"):
|
||||
object.__setattr__(self, "_assistant_reply_appended", False)
|
||||
|
||||
def _create_default_extension_state(self) -> ConversationState | None:
|
||||
initial_state_t = getattr(self, "_initial_state_t", None)
|
||||
@@ -852,6 +877,7 @@ class _ConversationalMixin:
|
||||
self._method_call_counts.clear()
|
||||
self._clear_or_listeners()
|
||||
self._is_execution_resuming = False
|
||||
object.__setattr__(self, "_turn_classified_intent", None)
|
||||
|
||||
def _apply_pending_conversational_turn(self) -> None:
|
||||
"""Drain the stashed user message + classify if intents configured.
|
||||
@@ -859,6 +885,7 @@ class _ConversationalMixin:
|
||||
Called from ``Flow.kickoff_async`` AFTER persist state restore so
|
||||
the appended message survives ``self.persistence.load_state(...)``.
|
||||
"""
|
||||
object.__setattr__(self, "_turn_classified_intent", None)
|
||||
if self._pending_user_message is None:
|
||||
return
|
||||
|
||||
@@ -1107,10 +1134,6 @@ class _ConversationalMixin:
|
||||
return "public"
|
||||
return "private"
|
||||
|
||||
def _assistant_message_count(self) -> int:
|
||||
state = cast(ConversationState, self.state)
|
||||
return sum(1 for message in state.messages if message.role == "assistant")
|
||||
|
||||
def _is_public_turn_result(self, result: Any) -> bool:
|
||||
if not isinstance(result, str):
|
||||
return False
|
||||
@@ -1190,6 +1213,15 @@ class _ConversationalMixin:
|
||||
)
|
||||
from crewai.events.types.flow_events import FlowFinishedEvent
|
||||
|
||||
# Background memory saves must finish (and emit their completed/failed
|
||||
# events) before the session-end flow_finished / batch finalization
|
||||
# below tears down listeners, mirroring the non-deferred kickoff path.
|
||||
# The flush then waits for those events' async bus handlers.
|
||||
drain_memory_writes = getattr(self, "_drain_memory_writes", None)
|
||||
if callable(drain_memory_writes):
|
||||
drain_memory_writes()
|
||||
crewai_event_bus.flush()
|
||||
|
||||
# Only emit the session-end event when a deferred flow_started is
|
||||
# actually pending. ``_deferred_flow_started_event_id`` is set only by
|
||||
# deferred kickoffs; when finalization was not deferred, each per-turn
|
||||
|
||||
@@ -956,6 +956,22 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
|
||||
return self.memory.remember_many(content, **kwargs)
|
||||
return self.memory.remember(content, **kwargs)
|
||||
|
||||
def _drain_memory_writes(self) -> None:
|
||||
"""Block until pending background memory saves for this flow finish.
|
||||
|
||||
Must run before ``FlowFinishedEvent`` is emitted: listeners (e.g.
|
||||
telemetry sessions) tear down on that event, and any
|
||||
``MemorySaveCompletedEvent``/``MemorySaveFailedEvent`` emitted after
|
||||
teardown is lost, leaving the save span orphaned.
|
||||
"""
|
||||
mem = self.memory
|
||||
if mem is None:
|
||||
return
|
||||
backing = getattr(mem, "_memory", None) or mem
|
||||
drain = getattr(backing, "drain_writes", None)
|
||||
if callable(drain):
|
||||
drain()
|
||||
|
||||
def extract_memories(self, content: str) -> list[str]:
|
||||
"""Extract discrete memories from content. Delegates to this flow's memory.
|
||||
|
||||
@@ -1474,6 +1490,14 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
|
||||
not self.suppress_flow_events
|
||||
and not self._should_defer_trace_finalization()
|
||||
):
|
||||
# Background memory saves must finish (and emit their
|
||||
# completed/failed events) before flow-finished triggers
|
||||
# listener teardown/finalization; the flush then waits for those
|
||||
# events' async handlers, mirroring Crew._create_crew_output.
|
||||
# Offloaded to a thread so the blocking waits don't stall other
|
||||
# coroutines on the loop.
|
||||
await asyncio.to_thread(self._drain_memory_writes)
|
||||
await asyncio.to_thread(crewai_event_bus.flush)
|
||||
future = crewai_event_bus.emit(
|
||||
self,
|
||||
FlowFinishedEvent(
|
||||
@@ -2285,6 +2309,14 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
|
||||
# flag is read from either the instance attribute or an extension
|
||||
# definition.
|
||||
if not self._should_defer_trace_finalization():
|
||||
# Background memory saves must finish (and emit their
|
||||
# completed/failed events) before flow-finished triggers
|
||||
# listener teardown/finalization; the flush then waits for
|
||||
# those events' async handlers, mirroring
|
||||
# Crew._create_crew_output. Offloaded to a thread so the
|
||||
# blocking waits don't stall other coroutines on the loop.
|
||||
await asyncio.to_thread(self._drain_memory_writes)
|
||||
await asyncio.to_thread(crewai_event_bus.flush)
|
||||
future = crewai_event_bus.emit(
|
||||
self,
|
||||
FlowFinishedEvent(
|
||||
@@ -2317,9 +2349,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
|
||||
|
||||
return final_output
|
||||
finally:
|
||||
# Ensure all background memory saves complete before returning
|
||||
if self.memory is not None and hasattr(self.memory, "drain_writes"):
|
||||
self.memory.drain_writes()
|
||||
# Safety net for the exception path; the success path already
|
||||
# drained before emitting FlowFinishedEvent.
|
||||
self._drain_memory_writes()
|
||||
# Drain pending LLMCallCompletedEvent handlers before
|
||||
# detaching so `flow.usage_metrics` reflects every call
|
||||
# emitted during this kickoff — mirrors `Crew.kickoff()`,
|
||||
|
||||
@@ -6,6 +6,7 @@ from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from crewai.types.usage_metrics import UsageMetrics
|
||||
from crewai.utilities.planning_types import TodoItem
|
||||
from crewai.utilities.types import LLMMessage
|
||||
|
||||
@@ -38,7 +39,13 @@ class LiteAgentOutput(BaseModel):
|
||||
)
|
||||
agent_role: str = Field(description="Role of the agent that produced this output")
|
||||
usage_metrics: dict[str, Any] | None = Field(
|
||||
description="Token usage metrics for this execution", default=None
|
||||
description=(
|
||||
"Token usage for this kickoff call only (guardrail retries "
|
||||
"included), not the LLM instance's cumulative totals, as a "
|
||||
"plain dict; ``token_usage`` exposes the same data as a "
|
||||
"UsageMetrics object"
|
||||
),
|
||||
default=None,
|
||||
)
|
||||
messages: list[LLMMessage] = Field(
|
||||
description="Messages of the agent", default_factory=list
|
||||
@@ -86,6 +93,19 @@ class LiteAgentOutput(BaseModel):
|
||||
return self.pydantic.model_dump()
|
||||
return {}
|
||||
|
||||
@property
|
||||
def token_usage(self) -> UsageMetrics:
|
||||
"""Token usage as a ``UsageMetrics`` object.
|
||||
|
||||
Same attribute name and type as ``CrewOutput.token_usage``, so a
|
||||
usage accessor written for one result type works on both. Returns
|
||||
zeroed metrics when no usage was captured (``usage_metrics`` is
|
||||
``None``).
|
||||
"""
|
||||
if not self.usage_metrics:
|
||||
return UsageMetrics()
|
||||
return UsageMetrics.model_validate(self.usage_metrics)
|
||||
|
||||
@property
|
||||
def completed_todos(self) -> list[TodoExecutionResult]:
|
||||
"""Get only the completed todos."""
|
||||
|
||||
@@ -394,19 +394,35 @@ class LLM(BaseLLM):
|
||||
"""Factory method that routes to native SDK or falls back to LiteLLM.
|
||||
|
||||
Routing priority:
|
||||
1. If 'provider' kwarg is present, use that provider with constants
|
||||
2. If only 'model' kwarg, use constants to infer provider
|
||||
3. If "/" in model name:
|
||||
1. If ``custom_openai=True``, force the native OpenAI provider,
|
||||
overriding any explicit provider. A custom endpoint is required.
|
||||
2. If ``provider`` is present, use that provider.
|
||||
3. If "/" is in the model name:
|
||||
- Check if prefix is a native provider (openai/anthropic/azure/bedrock/gemini)
|
||||
- If yes, validate model against constants
|
||||
- If valid, route to native SDK; otherwise route to LiteLLM
|
||||
4. Otherwise, infer the provider from the model name.
|
||||
"""
|
||||
if not model or not isinstance(model, str):
|
||||
raise ValueError("Model must be a non-empty string")
|
||||
|
||||
custom_openai = bool(kwargs.pop("custom_openai", False))
|
||||
custom_openai_route = custom_openai
|
||||
explicit_provider = kwargs.get("provider")
|
||||
|
||||
if explicit_provider:
|
||||
if custom_openai:
|
||||
if not cls._has_custom_openai_endpoint(kwargs):
|
||||
raise ValueError(
|
||||
"custom_openai=True requires base_url, api_base, "
|
||||
"OPENAI_BASE_URL, or OPENAI_API_BASE"
|
||||
)
|
||||
provider = "openai"
|
||||
use_native = True
|
||||
prefix, separator, model_part = model.partition("/")
|
||||
model_string = (
|
||||
model_part if separator and prefix.lower() == "openai" else model
|
||||
)
|
||||
elif explicit_provider:
|
||||
provider = explicit_provider
|
||||
use_native = True
|
||||
model_string = model
|
||||
@@ -435,9 +451,17 @@ class LLM(BaseLLM):
|
||||
|
||||
canonical_provider = provider_mapping.get(prefix.lower())
|
||||
|
||||
if canonical_provider and cls._validate_model_in_constants(
|
||||
model_part, canonical_provider
|
||||
):
|
||||
valid_native_model = bool(
|
||||
canonical_provider
|
||||
and cls._validate_model_in_constants(model_part, canonical_provider)
|
||||
)
|
||||
custom_openai_route = bool(
|
||||
canonical_provider == "openai"
|
||||
and not valid_native_model
|
||||
and cls._has_custom_openai_base_url(kwargs)
|
||||
)
|
||||
|
||||
if canonical_provider and (valid_native_model or custom_openai_route):
|
||||
provider = canonical_provider
|
||||
use_native = True
|
||||
model_string = model_part
|
||||
@@ -455,6 +479,8 @@ class LLM(BaseLLM):
|
||||
try:
|
||||
# Remove 'provider' from kwargs if it exists to avoid duplicate keyword argument
|
||||
kwargs_copy = {k: v for k, v in kwargs.items() if k != "provider"}
|
||||
if custom_openai_route:
|
||||
kwargs_copy["custom_openai"] = True
|
||||
return cast(
|
||||
Self,
|
||||
native_class(model=model_string, provider=provider, **kwargs_copy),
|
||||
@@ -590,6 +616,20 @@ class LLM(BaseLLM):
|
||||
|
||||
return cls._matches_provider_pattern(model, provider)
|
||||
|
||||
@staticmethod
|
||||
def _has_custom_openai_base_url(kwargs: dict[str, Any]) -> bool:
|
||||
"""Return whether this call explicitly configures a custom endpoint."""
|
||||
return bool(kwargs.get("base_url") or kwargs.get("api_base"))
|
||||
|
||||
@classmethod
|
||||
def _has_custom_openai_endpoint(cls, kwargs: dict[str, Any]) -> bool:
|
||||
"""Return whether a custom endpoint is configured explicitly or by env."""
|
||||
return bool(
|
||||
cls._has_custom_openai_base_url(kwargs)
|
||||
or os.getenv("OPENAI_BASE_URL")
|
||||
or os.getenv("OPENAI_API_BASE")
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _infer_provider_from_model(cls, model: str) -> str:
|
||||
"""Infer the provider from the model name.
|
||||
|
||||
@@ -966,8 +966,15 @@ class BaseLLM(BaseModel, ABC):
|
||||
def get_token_usage_summary(self) -> UsageMetrics:
|
||||
"""Get summary of token usage for this LLM instance.
|
||||
|
||||
The counters are cumulative for the lifetime of this instance: they
|
||||
grow across every call made through it, including calls issued by
|
||||
different agents sharing the instance. For usage scoped to a single
|
||||
call, snapshot before and after and use
|
||||
``UsageMetrics.delta_since`` (agent kickoff results already report
|
||||
per-call usage this way).
|
||||
|
||||
Returns:
|
||||
Dictionary with token usage totals
|
||||
UsageMetrics with this instance's lifetime token usage totals.
|
||||
"""
|
||||
return UsageMetrics(**self._token_usage)
|
||||
|
||||
|
||||
@@ -232,6 +232,7 @@ class OpenAICompletion(BaseLLM):
|
||||
auto_chain: bool = False
|
||||
auto_chain_reasoning: bool = False
|
||||
api_base: str | None = None
|
||||
custom_openai: bool = False
|
||||
is_o1_model: bool = False
|
||||
is_gpt4_model: bool = False
|
||||
|
||||
@@ -245,6 +246,20 @@ class OpenAICompletion(BaseLLM):
|
||||
def _normalize_openai_fields(cls, data: Any) -> Any:
|
||||
if not isinstance(data, dict):
|
||||
return data
|
||||
if data.get("custom_openai"):
|
||||
custom_base_url = (
|
||||
data.get("base_url")
|
||||
or data.get("api_base")
|
||||
or os.getenv("OPENAI_BASE_URL")
|
||||
or os.getenv("OPENAI_API_BASE")
|
||||
)
|
||||
if not custom_base_url:
|
||||
raise ValueError(
|
||||
"custom_openai=True requires base_url, api_base, "
|
||||
"OPENAI_BASE_URL, or OPENAI_API_BASE"
|
||||
)
|
||||
if not data.get("base_url") and not data.get("api_base"):
|
||||
data["base_url"] = custom_base_url
|
||||
if not data.get("provider"):
|
||||
data["provider"] = "openai"
|
||||
data["api_key"] = data.get("api_key") or os.getenv("OPENAI_API_KEY")
|
||||
@@ -355,6 +370,15 @@ class OpenAICompletion(BaseLLM):
|
||||
config["seed"] = self.seed
|
||||
if self.reasoning_effort is not None:
|
||||
config["reasoning_effort"] = self.reasoning_effort
|
||||
if self.custom_openai:
|
||||
config["model"] = self.model
|
||||
config["custom_openai"] = True
|
||||
config["base_url"] = (
|
||||
self.base_url
|
||||
or self.api_base
|
||||
or os.getenv("OPENAI_BASE_URL")
|
||||
or os.getenv("OPENAI_API_BASE")
|
||||
)
|
||||
return config
|
||||
|
||||
def _get_client_params(self) -> dict[str, Any]:
|
||||
@@ -372,6 +396,7 @@ class OpenAICompletion(BaseLLM):
|
||||
"base_url": self.base_url
|
||||
or self.api_base
|
||||
or os.getenv("OPENAI_BASE_URL")
|
||||
or os.getenv("OPENAI_API_BASE")
|
||||
or None,
|
||||
"timeout": self.timeout,
|
||||
"max_retries": self.max_retries,
|
||||
|
||||
@@ -5,7 +5,6 @@ import asyncio
|
||||
from collections.abc import Awaitable, Callable
|
||||
import importlib
|
||||
from inspect import Parameter, signature
|
||||
import json
|
||||
import threading
|
||||
from typing import (
|
||||
Any,
|
||||
@@ -37,9 +36,9 @@ from crewai.tools.structured_tool import (
|
||||
_infer_result_schema_from_callable,
|
||||
_serialize_schema,
|
||||
build_schema_hint,
|
||||
format_description_for_llm,
|
||||
)
|
||||
from crewai.types.callback import SerializableCallable, _resolve_dotted_path
|
||||
from crewai.utilities.pydantic_schema_utils import generate_model_description
|
||||
from crewai.utilities.string_utils import sanitize_tool_name
|
||||
|
||||
|
||||
@@ -479,15 +478,27 @@ class BaseTool(BaseModel, ABC):
|
||||
f"{self.__class__.__name__}Schema", **fields
|
||||
)
|
||||
|
||||
@property
|
||||
def formatted_description(self) -> str:
|
||||
"""LLM-facing composite of name, argument schema, and description.
|
||||
|
||||
Use this when rendering the tool into a prompt; ``description``
|
||||
holds only the authored text.
|
||||
"""
|
||||
return format_description_for_llm(self.name, self.args_schema, self.description)
|
||||
|
||||
def _generate_description(self) -> None:
|
||||
"""Generate the tool description with a JSON schema for arguments."""
|
||||
schema = generate_model_description(self.args_schema)
|
||||
args_json = json.dumps(schema["json_schema"]["schema"], indent=2)
|
||||
self.description = (
|
||||
f"Tool Name: {sanitize_tool_name(self.name)}\n"
|
||||
f"Tool Arguments: {args_json}\n"
|
||||
f"Tool Description: {self.description}"
|
||||
)
|
||||
"""Deprecated hook kept for backward compatibility; does nothing.
|
||||
|
||||
Historically this rewrote the public ``description`` field at
|
||||
construction time into the LLM-facing composite (``Tool Name: …\\n
|
||||
Tool Arguments: …\\nTool Description: <authored>``). The authored
|
||||
``description`` is now preserved as written and the composite is
|
||||
exposed separately via :attr:`formatted_description`.
|
||||
|
||||
``model_post_init`` still calls this so subclasses that override it
|
||||
(e.g. adapters that customize the composite) keep working.
|
||||
"""
|
||||
|
||||
|
||||
_BASE_TOOL_CLS = BaseTool
|
||||
|
||||
@@ -4,6 +4,7 @@ import asyncio
|
||||
from collections.abc import Callable
|
||||
import inspect
|
||||
import json
|
||||
import re
|
||||
import textwrap
|
||||
from typing import TYPE_CHECKING, Annotated, Any, cast, get_type_hints
|
||||
import warnings
|
||||
@@ -21,7 +22,10 @@ from pydantic import (
|
||||
from typing_extensions import Self
|
||||
|
||||
from crewai.utilities.logger import Logger
|
||||
from crewai.utilities.pydantic_schema_utils import create_model_from_schema
|
||||
from crewai.utilities.pydantic_schema_utils import (
|
||||
create_model_from_schema,
|
||||
generate_model_description,
|
||||
)
|
||||
from crewai.utilities.string_utils import sanitize_tool_name
|
||||
|
||||
|
||||
@@ -108,6 +112,70 @@ def build_schema_hint(args_schema: type[BaseModel]) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
# Matches a description that IS a pre-composed LLM block (as written by
|
||||
# older versions into the field, and by adapters that still bake it in).
|
||||
# Anchored to the full three-line shape so authored prose that merely
|
||||
# mentions "Tool Description:" is never mistaken for a composite. Greedy
|
||||
# ``.*`` keeps only the text after the LAST marker, matching the historical
|
||||
# split behavior for nested pre-baked blocks.
|
||||
_COMPOSITE_DESCRIPTION_RE = re.compile(
|
||||
r"^Tool Name:.*\nTool Arguments:.*\nTool Description:\s*",
|
||||
re.DOTALL,
|
||||
)
|
||||
|
||||
|
||||
def strip_composite_description_prefix(description: str) -> str:
|
||||
"""Return the authored text from a pre-composed LLM description block.
|
||||
|
||||
Descriptions that don't start with the composite shape are returned
|
||||
unchanged.
|
||||
"""
|
||||
match = _COMPOSITE_DESCRIPTION_RE.match(description)
|
||||
if match:
|
||||
return description[match.end() :]
|
||||
return description
|
||||
|
||||
|
||||
def format_description_for_llm(
|
||||
name: str,
|
||||
args_schema: type[BaseModel] | None,
|
||||
description: str,
|
||||
) -> str:
|
||||
"""Compose the LLM-facing tool description.
|
||||
|
||||
Combines the tool name, its argument JSON schema, and the authored
|
||||
description into the prompt block agents see. The authored
|
||||
``description`` field itself is never mutated — prompt rendering calls
|
||||
this on demand.
|
||||
|
||||
Idempotent: if ``description`` already *is* a composed block (e.g. a
|
||||
tool deserialized from a checkpoint written by an older version, or an
|
||||
adapter that bakes the composite into the field), only the authored
|
||||
text after the marker is used. The check is anchored to the composite
|
||||
shape, so authored prose that merely mentions ``"Tool Description:"``
|
||||
passes through untouched.
|
||||
|
||||
Args:
|
||||
name: The tool name (sanitized for the prompt).
|
||||
args_schema: The tool's argument schema, if any.
|
||||
description: The authored tool description.
|
||||
|
||||
Returns:
|
||||
The composed, LLM-facing description block.
|
||||
"""
|
||||
description = strip_composite_description_prefix(description)
|
||||
if args_schema is not None:
|
||||
schema = generate_model_description(args_schema)
|
||||
args_json = json.dumps(schema["json_schema"]["schema"], indent=2)
|
||||
else:
|
||||
args_json = "{}"
|
||||
return (
|
||||
f"Tool Name: {sanitize_tool_name(name)}\n"
|
||||
f"Tool Arguments: {args_json}\n"
|
||||
f"Tool Description: {description}"
|
||||
)
|
||||
|
||||
|
||||
class ToolUsageLimitExceededError(Exception):
|
||||
"""Exception raised when a tool has reached its maximum usage limit."""
|
||||
|
||||
@@ -141,6 +209,15 @@ class CrewStructuredTool(BaseModel):
|
||||
_logger: Logger = PrivateAttr(default_factory=Logger)
|
||||
_original_tool: Any = PrivateAttr(default=None)
|
||||
|
||||
@property
|
||||
def formatted_description(self) -> str:
|
||||
"""LLM-facing composite of name, argument schema, and description.
|
||||
|
||||
Use this when rendering the tool into a prompt; ``description``
|
||||
holds only the authored text.
|
||||
"""
|
||||
return format_description_for_llm(self.name, self.args_schema, self.description)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_func(self) -> Self:
|
||||
if self.func is not None:
|
||||
|
||||
@@ -430,7 +430,7 @@ class ToolUsage:
|
||||
).format(
|
||||
error=e,
|
||||
tool=sanitize_tool_name(tool.name),
|
||||
tool_inputs=tool.description,
|
||||
tool_inputs=tool.formatted_description,
|
||||
)
|
||||
result = ToolUsageError(
|
||||
f"\n{error_message}.\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
|
||||
@@ -670,7 +670,7 @@ class ToolUsage:
|
||||
).format(
|
||||
error=e,
|
||||
tool=sanitize_tool_name(tool.name),
|
||||
tool_inputs=tool.description,
|
||||
tool_inputs=tool.formatted_description,
|
||||
)
|
||||
result = ToolUsageError(
|
||||
f"\n{error_message}.\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
|
||||
@@ -803,7 +803,7 @@ class ToolUsage:
|
||||
|
||||
def _render(self) -> str:
|
||||
"""Render the tool name and description in plain text."""
|
||||
descriptions = [tool.description for tool in self.tools]
|
||||
descriptions = [tool.formatted_description for tool in self.tools]
|
||||
return "\n--\n".join(descriptions)
|
||||
|
||||
def _function_calling(
|
||||
|
||||
@@ -76,6 +76,38 @@ class UsageMetrics(BaseModel):
|
||||
self.cache_creation_tokens += usage_metrics.cache_creation_tokens
|
||||
self.successful_requests += usage_metrics.successful_requests
|
||||
|
||||
def delta_since(self, baseline: Self) -> Self:
|
||||
"""Return the per-call usage accrued since ``baseline`` was captured.
|
||||
|
||||
Both objects must come from the same monotonically increasing
|
||||
accumulator (e.g. an LLM instance's lifetime counters). Differences
|
||||
are clamped at zero so a reset accumulator can't produce negative
|
||||
usage.
|
||||
|
||||
Args:
|
||||
baseline: A snapshot of the same accumulator taken earlier.
|
||||
|
||||
Returns:
|
||||
A new UsageMetrics with the field-wise difference.
|
||||
"""
|
||||
return type(self)(
|
||||
total_tokens=max(0, self.total_tokens - baseline.total_tokens),
|
||||
prompt_tokens=max(0, self.prompt_tokens - baseline.prompt_tokens),
|
||||
cached_prompt_tokens=max(
|
||||
0, self.cached_prompt_tokens - baseline.cached_prompt_tokens
|
||||
),
|
||||
completion_tokens=max(
|
||||
0, self.completion_tokens - baseline.completion_tokens
|
||||
),
|
||||
reasoning_tokens=max(0, self.reasoning_tokens - baseline.reasoning_tokens),
|
||||
cache_creation_tokens=max(
|
||||
0, self.cache_creation_tokens - baseline.cache_creation_tokens
|
||||
),
|
||||
successful_requests=max(
|
||||
0, self.successful_requests - baseline.successful_requests
|
||||
),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_provider_dict(cls, usage_data: dict[str, Any] | None) -> Self | None:
|
||||
"""Normalize a provider's raw usage dict into a ``UsageMetrics``.
|
||||
|
||||
@@ -27,7 +27,10 @@ from crewai.agents.parser import (
|
||||
from crewai.llms.base_llm import BaseLLM, call_stop_override
|
||||
from crewai.tools import BaseTool as CrewAITool
|
||||
from crewai.tools.base_tool import BaseTool
|
||||
from crewai.tools.structured_tool import CrewStructuredTool
|
||||
from crewai.tools.structured_tool import (
|
||||
CrewStructuredTool,
|
||||
strip_composite_description_prefix,
|
||||
)
|
||||
from crewai.tools.tool_types import ToolResult
|
||||
from crewai.utilities.errors import AgentRepositoryError
|
||||
from crewai.utilities.exceptions.context_window_exceeding_exception import (
|
||||
@@ -147,7 +150,14 @@ def render_text_description_and_args(
|
||||
Returns:
|
||||
Plain text description of tools.
|
||||
"""
|
||||
tool_strings = [tool.description for tool in tools]
|
||||
# Fall back to the raw description for duck-typed tools (including test
|
||||
# mocks) that don't provide a real formatted_description string.
|
||||
tool_strings = [
|
||||
formatted
|
||||
if isinstance((formatted := getattr(tool, "formatted_description", None)), str)
|
||||
else tool.description
|
||||
for tool in tools
|
||||
]
|
||||
return "\n".join(tool_strings)
|
||||
|
||||
|
||||
@@ -190,10 +200,10 @@ def convert_tools_to_openai_schema(
|
||||
except Exception:
|
||||
parameters = {}
|
||||
|
||||
# BaseTool formats description as "Tool Name: ...\nTool Arguments: ...\nTool Description: {original}"
|
||||
description = tool.description
|
||||
if "Tool Description:" in description:
|
||||
description = description.split("Tool Description:")[-1].strip()
|
||||
# Old checkpoints and some adapters bake the composed LLM block
|
||||
# ("Tool Name: ...\nTool Arguments: ...\nTool Description: {authored}")
|
||||
# into the description field; keep only the authored text here.
|
||||
description = strip_composite_description_prefix(tool.description)
|
||||
|
||||
sanitized_name = sanitize_tool_name(tool.name)
|
||||
|
||||
|
||||
@@ -1138,3 +1138,160 @@ def test_lite_agent_memory_instance_recall_and_save_called():
|
||||
mock_memory.remember_many.assert_called_once_with(
|
||||
["Fact one.", "Fact two."], agent_role="Test"
|
||||
)
|
||||
|
||||
|
||||
class _FixedUsageLLM(BaseLLM):
|
||||
"""Offline BaseLLM that records fixed usage (100/10 tokens) per call."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(model="fixed-usage-model")
|
||||
|
||||
def call(
|
||||
self,
|
||||
messages,
|
||||
tools=None,
|
||||
callbacks=None,
|
||||
available_functions=None,
|
||||
from_task=None,
|
||||
from_agent=None,
|
||||
response_model=None,
|
||||
) -> str:
|
||||
self._track_token_usage_internal(
|
||||
{"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110}
|
||||
)
|
||||
return "Thought: I know the answer.\nFinal Answer: fake answer"
|
||||
|
||||
def supports_function_calling(self) -> bool:
|
||||
return False
|
||||
|
||||
def supports_stop_words(self) -> bool:
|
||||
return False
|
||||
|
||||
def get_context_window_size(self) -> int:
|
||||
return 4096
|
||||
|
||||
|
||||
class TestKickoffUsageMetricsArePerCall:
|
||||
"""Regression tests for EPD-177: kickoff results used to expose the LLM
|
||||
instance's cumulative lifetime counters, so counts accumulated across
|
||||
calls and pooled across agents sharing one LLM object.
|
||||
"""
|
||||
|
||||
def _make_agent(self, role: str, llm: BaseLLM) -> Agent:
|
||||
return Agent(
|
||||
role=role,
|
||||
goal="Answer questions.",
|
||||
backstory="Test agent.",
|
||||
llm=llm,
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
def test_agents_sharing_one_llm_report_per_call_usage(self):
|
||||
shared = _FixedUsageLLM()
|
||||
r1 = self._make_agent("agent one", shared).kickoff("question one")
|
||||
r2 = self._make_agent("agent two", shared).kickoff("question two")
|
||||
|
||||
assert r1.usage_metrics is not None
|
||||
assert r1.usage_metrics["prompt_tokens"] > 0
|
||||
# The second agent's call must not include the first agent's tokens.
|
||||
assert r2.usage_metrics == r1.usage_metrics
|
||||
|
||||
# The shared LLM instance still exposes cumulative lifetime totals.
|
||||
lifetime = shared.get_token_usage_summary()
|
||||
assert lifetime.prompt_tokens == (
|
||||
r1.usage_metrics["prompt_tokens"] + r2.usage_metrics["prompt_tokens"]
|
||||
)
|
||||
assert lifetime.successful_requests == (
|
||||
r1.usage_metrics["successful_requests"]
|
||||
+ r2.usage_metrics["successful_requests"]
|
||||
)
|
||||
|
||||
def test_repeated_kickoffs_on_same_agent_report_per_call_usage(self):
|
||||
agent = self._make_agent("agent", _FixedUsageLLM())
|
||||
r1 = agent.kickoff("question one")
|
||||
r2 = agent.kickoff("question two")
|
||||
|
||||
assert r1.usage_metrics is not None
|
||||
assert r1.usage_metrics["prompt_tokens"] > 0
|
||||
assert r2.usage_metrics == r1.usage_metrics
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_kickoff_reports_per_call_usage(self):
|
||||
shared = _FixedUsageLLM()
|
||||
r1 = await self._make_agent("agent one", shared).kickoff_async("question one")
|
||||
r2 = await self._make_agent("agent two", shared).kickoff_async("question two")
|
||||
|
||||
assert r1.usage_metrics is not None
|
||||
assert r1.usage_metrics["prompt_tokens"] > 0
|
||||
assert r2.usage_metrics == r1.usage_metrics
|
||||
|
||||
def test_guardrail_retry_usage_includes_all_attempts(self):
|
||||
"""A guardrail retry re-invokes the LLM within the same kickoff, so
|
||||
the result must report the whole call's usage — every attempt — not
|
||||
just the last one."""
|
||||
baseline = (
|
||||
self._make_agent("baseline", _FixedUsageLLM())
|
||||
.kickoff("question one")
|
||||
.usage_metrics
|
||||
)
|
||||
|
||||
attempts: list[str] = []
|
||||
|
||||
def flaky_guardrail(output):
|
||||
attempts.append(output.raw)
|
||||
if len(attempts) == 1:
|
||||
return (False, "Please try again.")
|
||||
return (True, output.raw)
|
||||
|
||||
agent = Agent(
|
||||
role="agent",
|
||||
goal="Answer questions.",
|
||||
backstory="Test agent.",
|
||||
llm=_FixedUsageLLM(),
|
||||
guardrail=flaky_guardrail,
|
||||
verbose=False,
|
||||
)
|
||||
result = agent.kickoff("question one")
|
||||
|
||||
assert len(attempts) == 2
|
||||
assert result.usage_metrics["successful_requests"] == (
|
||||
2 * baseline["successful_requests"]
|
||||
)
|
||||
assert result.usage_metrics["prompt_tokens"] == 2 * baseline["prompt_tokens"]
|
||||
assert result.usage_metrics["total_tokens"] == 2 * baseline["total_tokens"]
|
||||
|
||||
|
||||
class TestUsageMetricsDeltaSince:
|
||||
def test_field_wise_difference(self):
|
||||
baseline = UsageMetrics(
|
||||
total_tokens=110,
|
||||
prompt_tokens=100,
|
||||
completion_tokens=10,
|
||||
successful_requests=1,
|
||||
)
|
||||
current = UsageMetrics(
|
||||
total_tokens=330,
|
||||
prompt_tokens=300,
|
||||
completion_tokens=30,
|
||||
cached_prompt_tokens=5,
|
||||
reasoning_tokens=7,
|
||||
cache_creation_tokens=3,
|
||||
successful_requests=3,
|
||||
)
|
||||
|
||||
delta = current.delta_since(baseline)
|
||||
|
||||
assert delta == UsageMetrics(
|
||||
total_tokens=220,
|
||||
prompt_tokens=200,
|
||||
completion_tokens=20,
|
||||
cached_prompt_tokens=5,
|
||||
reasoning_tokens=7,
|
||||
cache_creation_tokens=3,
|
||||
successful_requests=2,
|
||||
)
|
||||
|
||||
def test_clamps_negative_differences_to_zero(self):
|
||||
baseline = UsageMetrics(total_tokens=100, prompt_tokens=90, successful_requests=2)
|
||||
delta = UsageMetrics().delta_since(baseline)
|
||||
assert delta == UsageMetrics()
|
||||
|
||||
@@ -30,10 +30,156 @@ def test_openai_completion_is_used_when_no_provider_prefix():
|
||||
llm = LLM(model="gpt-4o")
|
||||
|
||||
from crewai.llms.providers.openai.completion import OpenAICompletion
|
||||
assert isinstance(llm, OpenAICompletion)
|
||||
assert llm.__class__.__name__ == "OpenAICompletion"
|
||||
assert llm.provider == "openai"
|
||||
assert llm.model == "gpt-4o"
|
||||
|
||||
|
||||
def test_custom_openai_flag_uses_native_openai_without_provider_prefix():
|
||||
"""Custom OpenAI-compatible endpoints can serve arbitrary model ids."""
|
||||
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
|
||||
llm = LLM(
|
||||
model="anthropic/claude-sonnet-4-6",
|
||||
custom_openai=True,
|
||||
base_url="https://gateway.example/v1",
|
||||
is_litellm=False,
|
||||
)
|
||||
|
||||
assert llm.__class__.__name__ == "OpenAICompletion"
|
||||
assert llm.is_litellm is False
|
||||
assert llm.provider == "openai"
|
||||
assert llm.model == "anthropic/claude-sonnet-4-6"
|
||||
assert llm.base_url == "https://gateway.example/v1"
|
||||
assert llm.custom_openai is True
|
||||
assert "custom_openai" not in llm.additional_params
|
||||
|
||||
config = llm.to_config_dict()
|
||||
assert config["model"] == "anthropic/claude-sonnet-4-6"
|
||||
assert config["custom_openai"] is True
|
||||
assert config["base_url"] == "https://gateway.example/v1"
|
||||
|
||||
rebuilt = LLM(**config)
|
||||
assert isinstance(rebuilt, OpenAICompletion)
|
||||
assert rebuilt.model == "anthropic/claude-sonnet-4-6"
|
||||
assert rebuilt.base_url == "https://gateway.example/v1"
|
||||
|
||||
|
||||
def test_custom_openai_flag_requires_custom_base_url():
|
||||
"""Avoid routing arbitrary custom model ids to api.openai.com by mistake."""
|
||||
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=True):
|
||||
with pytest.raises(ValueError, match="custom_openai=True requires"):
|
||||
LLM(
|
||||
model="anthropic/claude-sonnet-4-6",
|
||||
custom_openai=True,
|
||||
is_litellm=False,
|
||||
)
|
||||
|
||||
|
||||
def test_direct_custom_openai_completion_requires_custom_base_url():
|
||||
"""Direct construction must not silently fall back to api.openai.com."""
|
||||
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=True):
|
||||
with pytest.raises(ValueError, match="custom_openai=True requires"):
|
||||
OpenAICompletion(
|
||||
model="anthropic/claude-sonnet-4-6",
|
||||
custom_openai=True,
|
||||
)
|
||||
|
||||
|
||||
def test_custom_openai_flag_strips_openai_routing_prefix():
|
||||
"""The openai/ routing prefix is not part of the gateway's model id."""
|
||||
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
|
||||
llm = LLM(
|
||||
model="openai/anthropic/claude-sonnet-4-6",
|
||||
custom_openai=True,
|
||||
base_url="https://gateway.example/v1",
|
||||
is_litellm=False,
|
||||
)
|
||||
|
||||
assert isinstance(llm, OpenAICompletion)
|
||||
assert llm.model == "anthropic/claude-sonnet-4-6"
|
||||
|
||||
|
||||
def test_openai_prefixed_custom_endpoint_uses_native_sdk_for_nested_model_id():
|
||||
"""Custom OpenAI-compatible endpoints may serve non-OpenAI model ids."""
|
||||
with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
|
||||
llm = LLM(
|
||||
model="openai/anthropic/claude-sonnet-4-6",
|
||||
base_url="https://gateway.example/v1",
|
||||
is_litellm=False,
|
||||
)
|
||||
|
||||
assert llm.__class__.__name__ == "OpenAICompletion"
|
||||
assert llm.is_litellm is False
|
||||
assert llm.provider == "openai"
|
||||
assert llm.model == "anthropic/claude-sonnet-4-6"
|
||||
assert llm.custom_openai is True
|
||||
assert llm.base_url == "https://gateway.example/v1"
|
||||
|
||||
def test_explicit_custom_openai_uses_legacy_api_base_env_var():
|
||||
"""Explicit custom routing supports the legacy endpoint environment variable."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"OPENAI_API_KEY": "test-key",
|
||||
"OPENAI_API_BASE": "https://gateway.example/v1",
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
os.environ.pop("OPENAI_BASE_URL", None)
|
||||
llm = LLM(
|
||||
model="openai/anthropic/claude-sonnet-4-6",
|
||||
custom_openai=True,
|
||||
is_litellm=False,
|
||||
)
|
||||
|
||||
assert isinstance(llm, OpenAICompletion)
|
||||
assert llm.is_litellm is False
|
||||
assert llm.provider == "openai"
|
||||
assert llm.model == "anthropic/claude-sonnet-4-6"
|
||||
assert llm.custom_openai is True
|
||||
|
||||
|
||||
def test_openai_prefixed_unknown_model_ignores_ambient_base_url_for_routing():
|
||||
"""Ambient OpenAI configuration must not opt unknown models into native routing."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"OPENAI_API_KEY": "test-key",
|
||||
"OPENAI_BASE_URL": "https://gateway.example/v1",
|
||||
},
|
||||
clear=True,
|
||||
):
|
||||
with (
|
||||
patch("crewai.llm._ensure_litellm", return_value=False),
|
||||
pytest.raises(ImportError, match="LiteLLM fallback package"),
|
||||
):
|
||||
LLM(model="openai/not-a-real-openai-model")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("endpoint_field", ["api_base", "env"])
|
||||
def test_custom_openai_config_preserves_resolved_endpoint(endpoint_field):
|
||||
"""Serialized custom OpenAI configs can reconstruct the same endpoint."""
|
||||
kwargs = {}
|
||||
env = {"OPENAI_API_KEY": "test-key"}
|
||||
if endpoint_field == "api_base":
|
||||
kwargs["api_base"] = "https://gateway.example/v1"
|
||||
else:
|
||||
env["OPENAI_API_BASE"] = "https://gateway.example/v1"
|
||||
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
llm = LLM(
|
||||
model="anthropic/claude-sonnet-4-6",
|
||||
custom_openai=True,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
config = llm.to_config_dict()
|
||||
assert config["base_url"] == "https://gateway.example/v1"
|
||||
rebuilt = LLM(**config)
|
||||
assert isinstance(rebuilt, OpenAICompletion)
|
||||
assert rebuilt.base_url == "https://gateway.example/v1"
|
||||
|
||||
|
||||
@pytest.mark.vcr()
|
||||
def test_openai_is_default_provider_without_explicit_llm_set_on_agent():
|
||||
"""
|
||||
@@ -60,14 +206,13 @@ def test_openai_is_default_provider_without_explicit_llm_set_on_agent():
|
||||
|
||||
|
||||
|
||||
def test_openai_completion_module_is_imported():
|
||||
def test_openai_completion_module_is_imported(monkeypatch):
|
||||
"""
|
||||
Test that the completion module is properly imported when using OpenAI provider
|
||||
"""
|
||||
module_name = "crewai.llms.providers.openai.completion"
|
||||
|
||||
if module_name in sys.modules:
|
||||
del sys.modules[module_name]
|
||||
monkeypatch.delitem(sys.modules, module_name, raising=False)
|
||||
|
||||
LLM(model="gpt-4o")
|
||||
|
||||
@@ -421,12 +566,25 @@ def test_openai_get_client_params_with_env_var():
|
||||
client_params = llm._get_client_params()
|
||||
assert client_params["base_url"] == "https://env.openai.com/v1"
|
||||
|
||||
def test_openai_get_client_params_with_legacy_api_base_env_var():
|
||||
"""
|
||||
Test that _get_client_params uses OPENAI_API_BASE when OPENAI_BASE_URL is absent.
|
||||
"""
|
||||
with patch.dict(os.environ, {
|
||||
"OPENAI_API_BASE": "https://legacy-env.openai.com/v1",
|
||||
}, clear=False):
|
||||
os.environ.pop("OPENAI_BASE_URL", None)
|
||||
llm = OpenAICompletion(model="gpt-4o")
|
||||
client_params = llm._get_client_params()
|
||||
assert client_params["base_url"] == "https://legacy-env.openai.com/v1"
|
||||
|
||||
def test_openai_get_client_params_priority_order():
|
||||
"""
|
||||
Test the priority order: base_url > api_base > OPENAI_BASE_URL env var
|
||||
Test the priority order: base_url > api_base > OPENAI_BASE_URL > OPENAI_API_BASE
|
||||
"""
|
||||
with patch.dict(os.environ, {
|
||||
"OPENAI_BASE_URL": "https://env.openai.com/v1",
|
||||
"OPENAI_API_BASE": "https://legacy-env.openai.com/v1",
|
||||
}):
|
||||
llm1 = OpenAICompletion(
|
||||
model="gpt-4o",
|
||||
|
||||
@@ -859,6 +859,7 @@ def test_cache_hitting_between_agents(researcher, writer, ceo):
|
||||
crew = Crew(
|
||||
agents=[ceo, researcher],
|
||||
tasks=tasks,
|
||||
cache=True,
|
||||
)
|
||||
|
||||
with patch.object(CacheHandler, "read") as read:
|
||||
@@ -2246,7 +2247,9 @@ def test_tools_with_custom_caching():
|
||||
agent=writer2,
|
||||
)
|
||||
|
||||
crew = Crew(agents=[writer1, writer2], tasks=[task1, task2, task3, task4])
|
||||
crew = Crew(
|
||||
agents=[writer1, writer2], tasks=[task1, task2, task3, task4], cache=True
|
||||
)
|
||||
|
||||
with patch.object(
|
||||
CacheHandler, "add", wraps=crew._cache_handler.add
|
||||
@@ -4598,6 +4601,98 @@ def test_reset_memory_uses_full_unified_memory_reset(researcher):
|
||||
reset.assert_not_called()
|
||||
|
||||
|
||||
def test_kickoff_drains_pending_memory_saves_before_completion_event(researcher):
|
||||
"""Background memory saves must finish (and emit their completion events)
|
||||
before CrewKickoffCompletedEvent, otherwise listeners that tear down on
|
||||
kickoff-completed (e.g. telemetry sessions) see the save span as orphaned."""
|
||||
import time
|
||||
|
||||
from crewai.events.types.crew_events import CrewKickoffCompletedEvent
|
||||
|
||||
order: list[str] = []
|
||||
|
||||
def slow_save():
|
||||
time.sleep(0.3)
|
||||
order.append("save-done")
|
||||
|
||||
crew = Crew(
|
||||
agents=[researcher],
|
||||
process=Process.sequential,
|
||||
tasks=[
|
||||
Task(description="Task 1", expected_output="output", agent=researcher),
|
||||
],
|
||||
memory=True,
|
||||
task_callback=lambda _output: crew._memory._submit_save(slow_save),
|
||||
)
|
||||
|
||||
completed = threading.Event()
|
||||
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(CrewKickoffCompletedEvent)
|
||||
def on_completed(_source, _event):
|
||||
order.append("kickoff-completed")
|
||||
completed.set()
|
||||
|
||||
with patch.object(Agent, "execute_task", return_value="ok"):
|
||||
crew.kickoff()
|
||||
|
||||
assert completed.wait(timeout=5)
|
||||
|
||||
assert order.index("save-done") < order.index("kickoff-completed")
|
||||
|
||||
|
||||
def test_kickoff_drains_agent_memory_saves_before_completion_event(tmp_path):
|
||||
"""Agents save through their own ``agent.memory`` when set; those pools
|
||||
must also be drained before CrewKickoffCompletedEvent."""
|
||||
import time
|
||||
|
||||
from crewai.events.types.crew_events import CrewKickoffCompletedEvent
|
||||
|
||||
agent_memory = Memory(storage=str(tmp_path / "agent-mem"))
|
||||
agent_with_memory = Agent(
|
||||
role="Researcher",
|
||||
goal="Research things",
|
||||
backstory="Experienced researcher",
|
||||
memory=agent_memory,
|
||||
)
|
||||
|
||||
order: list[str] = []
|
||||
|
||||
def slow_save():
|
||||
time.sleep(0.3)
|
||||
order.append("save-done")
|
||||
|
||||
crew = Crew(
|
||||
agents=[agent_with_memory],
|
||||
process=Process.sequential,
|
||||
tasks=[
|
||||
Task(
|
||||
description="Task 1",
|
||||
expected_output="output",
|
||||
agent=agent_with_memory,
|
||||
),
|
||||
],
|
||||
task_callback=lambda _output: agent_memory._submit_save(slow_save),
|
||||
)
|
||||
|
||||
completed = threading.Event()
|
||||
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(CrewKickoffCompletedEvent)
|
||||
def on_completed(_source, _event):
|
||||
order.append("kickoff-completed")
|
||||
completed.set()
|
||||
|
||||
with patch.object(Agent, "execute_task", return_value="ok"):
|
||||
crew.kickoff()
|
||||
|
||||
assert completed.wait(timeout=5)
|
||||
|
||||
assert order.index("save-done") < order.index("kickoff-completed")
|
||||
|
||||
|
||||
def test_reset_knowledge_with_only_crew_knowledge(researcher, writer):
|
||||
mock_ks = MagicMock(spec=Knowledge)
|
||||
|
||||
|
||||
@@ -2353,3 +2353,41 @@ def test_locked_dict_proxy_ior():
|
||||
def test_locked_dict_proxy_reversed():
|
||||
flow = _make_dict_flow()
|
||||
assert list(reversed(flow.state.data)) == ["c", "b", "a"]
|
||||
|
||||
|
||||
def test_flow_drains_pending_memory_saves_before_finished_event(tmp_path):
|
||||
"""Background memory saves must finish (and emit their completion events)
|
||||
before FlowFinishedEvent, otherwise listeners that tear down on
|
||||
flow-finished (e.g. telemetry sessions) see the save span as orphaned."""
|
||||
import time
|
||||
|
||||
from crewai.memory.unified_memory import Memory
|
||||
|
||||
order: list[str] = []
|
||||
|
||||
def slow_save():
|
||||
time.sleep(0.3)
|
||||
order.append("save-done")
|
||||
|
||||
class MemoryFlow(Flow):
|
||||
@start()
|
||||
def step_1(self):
|
||||
self.memory._submit_save(slow_save)
|
||||
return "done"
|
||||
|
||||
flow = MemoryFlow(memory=Memory(storage=str(tmp_path / "flow-mem")))
|
||||
|
||||
finished = threading.Event()
|
||||
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(FlowFinishedEvent)
|
||||
def on_finished(_source, _event):
|
||||
order.append("flow-finished")
|
||||
finished.set()
|
||||
|
||||
flow.kickoff()
|
||||
|
||||
assert finished.wait(timeout=5)
|
||||
|
||||
assert order.index("save-done") < order.index("flow-finished")
|
||||
|
||||
@@ -1555,6 +1555,180 @@ class TestConversationalFlow:
|
||||
)
|
||||
|
||||
|
||||
class TestHandleTurnReplyFallback:
|
||||
"""Regression tests for EPD-181: ``handle_turn()`` decided "did the
|
||||
handler append its reply?" by comparing assistant-message counts. A
|
||||
handler that appends its reply AND trims history to a cap left the count
|
||||
unchanged, so the fallback appended the reply a second time — every turn,
|
||||
once trimming engaged. The check now uses an explicit appended-this-turn
|
||||
flag.
|
||||
"""
|
||||
|
||||
MAX_MESSAGES = 4
|
||||
|
||||
def _make_bot(self) -> ConversationalFlow:
|
||||
max_messages = self.MAX_MESSAGES
|
||||
|
||||
class EchoBot(ConversationalFlow):
|
||||
def route_turn(self, context: dict[str, Any]) -> str | None:
|
||||
return "ECHO"
|
||||
|
||||
@listen("ECHO")
|
||||
def echo(self) -> str:
|
||||
reply = f"echo: {self.state.current_user_message or ''}"
|
||||
self.append_assistant_message(reply) # handler DOES append
|
||||
if len(self.state.messages) > max_messages: # ...and trims
|
||||
self.state.messages = self.state.messages[-max_messages:]
|
||||
return reply
|
||||
|
||||
return EchoBot()
|
||||
|
||||
def test_no_duplicate_reply_when_handler_trims_history(self) -> None:
|
||||
bot = self._make_bot()
|
||||
for i in range(1, 5):
|
||||
bot.handle_turn(f"message {i}")
|
||||
contents = [message.content for message in bot.state.messages]
|
||||
assert len(contents) == len(set(contents)), (
|
||||
f"duplicate reply on turn {i}: {contents}"
|
||||
)
|
||||
|
||||
# The capped window holds the last two full turns, in order.
|
||||
assert [message.content for message in bot.state.messages] == [
|
||||
"message 3",
|
||||
"echo: message 3",
|
||||
"message 4",
|
||||
"echo: message 4",
|
||||
]
|
||||
|
||||
def test_fallback_still_appends_when_handler_does_not_reply(self) -> None:
|
||||
class SilentBot(ConversationalFlow):
|
||||
def route_turn(self, context: dict[str, Any]) -> str | None:
|
||||
return "WORK"
|
||||
|
||||
@listen("WORK")
|
||||
def work(self) -> str:
|
||||
return "computed reply" # returns without appending
|
||||
|
||||
bot = SilentBot()
|
||||
bot.handle_turn("hello")
|
||||
|
||||
assistant_messages = [
|
||||
message.content
|
||||
for message in bot.state.messages
|
||||
if message.role == "assistant"
|
||||
]
|
||||
assert assistant_messages == ["computed reply"]
|
||||
|
||||
|
||||
class TestFalsyRouteTurnFallback:
|
||||
"""A falsy ``route_turn()`` must never replay a previous turn's intent.
|
||||
|
||||
Regression tests for EPD-176: an overridden ``route_turn()`` returning
|
||||
``None`` on an unhandled input used to silently reuse the sticky
|
||||
``state.last_intent`` from the *previous* turn, running the wrong handler
|
||||
with no error or warning.
|
||||
"""
|
||||
|
||||
def test_falsy_route_turn_does_not_replay_previous_turns_intent(self) -> None:
|
||||
ran: list[str] = []
|
||||
|
||||
class Bot(ConversationalFlow):
|
||||
def route_turn(self, context: dict[str, Any]) -> str | None:
|
||||
message = context.get("current_user_message") or ""
|
||||
if "hello" in message.lower():
|
||||
return "GREETING"
|
||||
return None # unhandled input -> falsy return
|
||||
|
||||
@listen("GREETING")
|
||||
def greeting(self) -> str:
|
||||
ran.append("GREETING")
|
||||
reply = "Hi! I only do greetings."
|
||||
self.append_assistant_message(reply)
|
||||
return reply
|
||||
|
||||
@listen("WEATHER")
|
||||
def weather(self) -> str:
|
||||
ran.append("WEATHER")
|
||||
reply = "It is sunny."
|
||||
self.append_assistant_message(reply)
|
||||
return reply
|
||||
|
||||
flow = Bot()
|
||||
flow.handle_turn("hello there")
|
||||
assert ran == ["GREETING"]
|
||||
assert flow.state.last_intent == "GREETING"
|
||||
|
||||
flow.handle_turn("what is the meaning of life?")
|
||||
assert ran == ["GREETING"], (
|
||||
"an unhandled turn must not re-run the previous turn's handler"
|
||||
)
|
||||
# With no routing decision the turn falls through to the built-in
|
||||
# 'converse' default instead of replaying the stale intent.
|
||||
assert flow.state.last_intent == "converse"
|
||||
assert flow.state.messages[-1].content != "Hi! I only do greetings."
|
||||
|
||||
def test_stale_intent_ignored_but_route_selected_event_still_emitted(
|
||||
self,
|
||||
) -> None:
|
||||
class Bot(ConversationalFlow):
|
||||
def route_turn(self, context: dict[str, Any]) -> str | None:
|
||||
message = context.get("current_user_message") or ""
|
||||
return "work" if "work" in message else None
|
||||
|
||||
@listen("work")
|
||||
def do_work(self) -> str:
|
||||
self.append_assistant_message("worked")
|
||||
return "worked"
|
||||
|
||||
flow = Bot()
|
||||
routes: list[ConversationRouteSelectedEvent] = []
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(ConversationRouteSelectedEvent)
|
||||
def capture(_: Any, event: ConversationRouteSelectedEvent) -> None:
|
||||
routes.append(event)
|
||||
|
||||
flow.handle_turn("work please")
|
||||
flow.handle_turn("something unrelated")
|
||||
crewai_event_bus.flush()
|
||||
|
||||
assert [event.route for event in routes] == ["work", "converse"]
|
||||
# The fallback decision still reports the prior intent for visibility.
|
||||
assert routes[1].previous_intent == "work"
|
||||
|
||||
def test_fresh_intent_classified_this_turn_still_routes(self) -> None:
|
||||
"""The legacy ``default_intents`` path classifies per turn and must
|
||||
keep routing on the freshly classified intent — including when the
|
||||
intent changes between turns."""
|
||||
ran: list[str] = []
|
||||
|
||||
@ConversationConfig(
|
||||
default_intents=["search", "weather"], intent_llm="gpt-4o-mini"
|
||||
)
|
||||
class LegacyFlow(ConversationalFlow):
|
||||
@listen("search")
|
||||
def handle_search(self) -> str:
|
||||
ran.append("search")
|
||||
self.append_assistant_message("searched")
|
||||
return "searched"
|
||||
|
||||
@listen("weather")
|
||||
def handle_weather(self) -> str:
|
||||
ran.append("weather")
|
||||
self.append_assistant_message("sunny")
|
||||
return "sunny"
|
||||
|
||||
flow = LegacyFlow()
|
||||
with patch.object(
|
||||
flow, "_collapse_to_outcome", side_effect=["search", "weather"]
|
||||
):
|
||||
flow.handle_turn("look up crewai")
|
||||
flow.handle_turn("how is the weather?")
|
||||
|
||||
assert ran == ["search", "weather"]
|
||||
assert flow.state.last_intent == "weather"
|
||||
|
||||
|
||||
class TestFlowTracingWhenSuppressed:
|
||||
def test_flow_started_emitted_when_panel_events_suppressed(self) -> None:
|
||||
class QuietFlow(Flow[ChatState]):
|
||||
|
||||
263
lib/crewai/tests/test_tool_cache_default.py
Normal file
263
lib/crewai/tests/test_tool_cache_default.py
Normal file
@@ -0,0 +1,263 @@
|
||||
# mypy: ignore-errors
|
||||
"""Regression tests for EPD-180: tool-result caching used to be ON by default,
|
||||
so an LLM calling the same tool with identical arguments twice in one run got
|
||||
the first (possibly stale) result back without the tool executing — silently
|
||||
wrong for live-data tools, and silently dropped actions for stateful tools.
|
||||
|
||||
Caching is now opt-in: ``Crew(cache=True)`` for crews, ``Agent(cache=True)``
|
||||
(or an explicit ``cache_handler``) for standalone agents. The machinery —
|
||||
including per-tool ``cache_function`` write gating — is unchanged once opted
|
||||
in.
|
||||
|
||||
The end-to-end tests run fully offline: a fake OpenAI client scripts two
|
||||
identical tool calls followed by a final answer, mirroring the EPD-180
|
||||
clean-room repro.
|
||||
"""
|
||||
|
||||
from openai.types.chat import ChatCompletion
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from crewai import LLM, Agent, Crew, Task
|
||||
from crewai.agents.cache.cache_handler import CacheHandler
|
||||
from crewai.tools import BaseTool
|
||||
|
||||
|
||||
class LookupArgs(BaseModel):
|
||||
city: str = Field(description="City to look up.")
|
||||
|
||||
|
||||
def make_live_tool():
|
||||
"""A tool returning a different value on every real execution."""
|
||||
executions = []
|
||||
|
||||
class LiveLookupTool(BaseTool):
|
||||
name: str = "live_lookup"
|
||||
description: str = "Returns a live (time-varying) reading for a city."
|
||||
args_schema: type[BaseModel] = LookupArgs
|
||||
# cache_function deliberately NOT set — exercising the default.
|
||||
|
||||
def _run(self, city: str) -> str:
|
||||
executions.append(city)
|
||||
return f"reading #{len(executions)} for {city}"
|
||||
|
||||
return LiveLookupTool(), executions
|
||||
|
||||
|
||||
def make_scripted_llm():
|
||||
"""An offline LLM whose client scripts two identical tool calls."""
|
||||
|
||||
def tool_call_response(call_id: str):
|
||||
return {
|
||||
"index": 0,
|
||||
"finish_reason": "tool_calls",
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": call_id,
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "live_lookup",
|
||||
"arguments": '{"city": "paris"}',
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
scripted = [
|
||||
tool_call_response("call_1"),
|
||||
tool_call_response("call_2"), # identical name+args, new id
|
||||
{
|
||||
"index": 0,
|
||||
"finish_reason": "stop",
|
||||
"message": {"role": "assistant", "content": "Final answer: done."},
|
||||
},
|
||||
]
|
||||
|
||||
class FakeCompletions:
|
||||
def __init__(self):
|
||||
self.n = 0
|
||||
|
||||
def create(self, **params):
|
||||
choice = scripted[min(self.n, len(scripted) - 1)]
|
||||
self.n += 1
|
||||
return ChatCompletion.model_validate(
|
||||
{
|
||||
"id": f"chatcmpl-fake-{self.n}",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": params.get("model", "gpt-4o"),
|
||||
"choices": [choice],
|
||||
"usage": {
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 5,
|
||||
"total_tokens": 15,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self):
|
||||
self.chat = type("Chat", (), {"completions": FakeCompletions()})()
|
||||
|
||||
llm = LLM(model="openai/gpt-4o")
|
||||
llm._client = FakeClient()
|
||||
return llm
|
||||
|
||||
|
||||
def run_crew(**crew_kwargs):
|
||||
tool, executions = make_live_tool()
|
||||
agent = Agent(
|
||||
role="reader",
|
||||
goal="Look things up.",
|
||||
backstory="Test agent.",
|
||||
llm=make_scripted_llm(),
|
||||
tools=[tool],
|
||||
verbose=False,
|
||||
)
|
||||
task = Task(
|
||||
description="Look up paris twice and report.",
|
||||
expected_output="A report.",
|
||||
agent=agent,
|
||||
)
|
||||
crew = Crew(agents=[agent], tasks=[task], verbose=False, **crew_kwargs)
|
||||
crew.kickoff()
|
||||
return executions
|
||||
|
||||
|
||||
class TestToolCachingIsOptIn:
|
||||
def test_default_reexecutes_identical_tool_calls(self):
|
||||
"""EPD-180: with no opt-in, both identical calls must really execute."""
|
||||
executions = run_crew()
|
||||
assert len(executions) == 2
|
||||
|
||||
def test_crew_cache_true_dedupes_identical_tool_calls(self):
|
||||
"""Opting in via Crew(cache=True) restores the dedup behavior."""
|
||||
executions = run_crew(cache=True)
|
||||
assert len(executions) == 1
|
||||
|
||||
|
||||
class TestAgentCacheWiring:
|
||||
def _agent(self, **kwargs) -> Agent:
|
||||
return Agent(
|
||||
role="reader",
|
||||
goal="Look things up.",
|
||||
backstory="Test agent.",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def test_standalone_agent_has_no_cache_by_default(self):
|
||||
agent = self._agent()
|
||||
assert agent.tools_handler.cache is None
|
||||
assert agent.cache_handler is None
|
||||
|
||||
def test_standalone_agent_explicit_cache_true_opts_in(self):
|
||||
agent = self._agent(cache=True)
|
||||
assert agent.tools_handler.cache is not None
|
||||
assert agent.cache_handler is not None
|
||||
|
||||
def test_standalone_agent_explicit_cache_handler_opts_in(self):
|
||||
handler = CacheHandler()
|
||||
agent = self._agent(cache_handler=handler)
|
||||
assert agent.tools_handler.cache is handler
|
||||
|
||||
def test_explicit_cache_false_stays_off_even_with_handler(self):
|
||||
agent = self._agent(cache=False, cache_handler=CacheHandler())
|
||||
assert agent.tools_handler.cache is None
|
||||
|
||||
def test_agents_accept_a_crew_offered_handler_by_default(self):
|
||||
"""``Crew(cache=True)`` offers its handler via set_cache_handler at
|
||||
kickoff; agents that didn't explicitly opt out must accept it."""
|
||||
agent = self._agent()
|
||||
assert agent.tools_handler.cache is None
|
||||
|
||||
handler = CacheHandler()
|
||||
agent.set_cache_handler(handler)
|
||||
assert agent.tools_handler.cache is handler
|
||||
|
||||
def test_agents_that_opted_out_refuse_a_crew_offered_handler(self):
|
||||
agent = self._agent(cache=False)
|
||||
agent.set_cache_handler(CacheHandler())
|
||||
assert agent.tools_handler.cache is None
|
||||
|
||||
def test_copy_of_default_agent_does_not_opt_in(self):
|
||||
"""copy() rebuilds from model_dump(), which includes the field
|
||||
default cache=True — that must not read as an explicit opt-in on
|
||||
the copy (Bugbot review finding on the original PR)."""
|
||||
copied = self._agent().copy()
|
||||
assert copied.tools_handler.cache is None
|
||||
assert copied.cache_handler is None
|
||||
|
||||
def test_copy_of_opted_in_agent_stays_opted_in(self):
|
||||
copied = self._agent(cache=True).copy()
|
||||
assert copied.tools_handler.cache is not None
|
||||
|
||||
def test_copy_of_handler_opted_in_agent_stays_opted_in(self):
|
||||
"""An explicit cache_handler is an opt-in too; copy() excludes the
|
||||
handler itself, but the consent must survive — the copy wires its
|
||||
own fresh handler (Bugbot review finding on the original PR)."""
|
||||
source = self._agent(cache_handler=CacheHandler())
|
||||
copied = source.copy()
|
||||
assert copied.tools_handler.cache is not None
|
||||
assert copied.tools_handler.cache is not source.tools_handler.cache
|
||||
|
||||
def test_copy_of_explicit_cache_false_with_handler_stays_off(self):
|
||||
copied = self._agent(cache=False, cache_handler=CacheHandler()).copy()
|
||||
assert copied.tools_handler.cache is None
|
||||
|
||||
def test_copy_of_crew_wired_agent_does_not_opt_in(self):
|
||||
"""A handler offered by a crew at kickoff (set_cache_handler) is
|
||||
runtime wiring, not construction-time consent — copies of such
|
||||
agents must not become standalone cachers (Bugbot review finding
|
||||
on the original PR)."""
|
||||
agent = self._agent()
|
||||
agent.set_cache_handler(CacheHandler()) # what Crew(cache=True) does
|
||||
assert agent.tools_handler.cache is not None
|
||||
|
||||
copied = agent.copy()
|
||||
assert copied.tools_handler.cache is None
|
||||
assert copied.cache_handler is None
|
||||
|
||||
|
||||
class TestHierarchicalManagerCacheWiring:
|
||||
"""The auto-created hierarchical manager is built outside the agents
|
||||
loop that offers the crew's cache handler; an opted-in crew must wire
|
||||
the manager too (Bugbot review finding on the original PR)."""
|
||||
|
||||
def _crew(self, **crew_kwargs) -> Crew:
|
||||
from crewai.process import Process
|
||||
|
||||
agent = Agent(role="worker", goal="Do work.", backstory="Test agent.")
|
||||
task = Task(description="Do the work.", expected_output="Done.")
|
||||
return Crew(
|
||||
agents=[agent],
|
||||
tasks=[task],
|
||||
process=Process.hierarchical,
|
||||
manager_llm="gpt-4o",
|
||||
**crew_kwargs,
|
||||
)
|
||||
|
||||
def test_manager_gets_crew_handler_when_cache_enabled(self):
|
||||
crew = self._crew(cache=True)
|
||||
crew._create_manager_agent()
|
||||
assert crew.manager_agent.tools_handler.cache is crew._cache_handler
|
||||
|
||||
def test_manager_has_no_cache_when_crew_did_not_opt_in(self):
|
||||
crew = self._crew()
|
||||
crew._create_manager_agent()
|
||||
assert crew.manager_agent.tools_handler.cache is None
|
||||
|
||||
def test_user_provided_manager_with_cache_false_stays_excluded(self):
|
||||
manager = Agent(
|
||||
role="manager",
|
||||
goal="Manage.",
|
||||
backstory="Test manager.",
|
||||
cache=False,
|
||||
allow_delegation=True,
|
||||
)
|
||||
crew = self._crew(cache=True)
|
||||
crew.manager_agent = manager
|
||||
crew._create_manager_agent()
|
||||
assert manager.tools_handler.cache is None
|
||||
143
lib/crewai/tests/test_usage_shape_parity.py
Normal file
143
lib/crewai/tests/test_usage_shape_parity.py
Normal file
@@ -0,0 +1,143 @@
|
||||
# mypy: ignore-errors
|
||||
"""Regression tests for EPD-178: token usage was exposed in different shapes
|
||||
and attribute names per code path — ``Agent.kickoff()`` results carried a
|
||||
plain dict at ``.usage_metrics`` (no ``token_usage`` attribute at all), while
|
||||
``Crew.kickoff()`` results carried a ``UsageMetrics`` object at
|
||||
``.token_usage`` (no ``usage_metrics`` attribute), so any single accessor
|
||||
written for one path raised ``AttributeError`` on the other.
|
||||
|
||||
Both result types now expose both surfaces: ``.token_usage`` as a
|
||||
``UsageMetrics`` object and ``.usage_metrics`` as a plain dict.
|
||||
"""
|
||||
|
||||
from crewai import Agent, Crew, Task
|
||||
from crewai.crews.crew_output import CrewOutput
|
||||
from crewai.lite_agent_output import LiteAgentOutput
|
||||
from crewai.llms.base_llm import BaseLLM
|
||||
from crewai.types.usage_metrics import UsageMetrics
|
||||
|
||||
|
||||
class _FixedUsageLLM(BaseLLM):
|
||||
"""Offline BaseLLM that records fixed usage (100/10 tokens) per call."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(model="fixed-usage-model")
|
||||
|
||||
def call(
|
||||
self,
|
||||
messages,
|
||||
tools=None,
|
||||
callbacks=None,
|
||||
available_functions=None,
|
||||
from_task=None,
|
||||
from_agent=None,
|
||||
response_model=None,
|
||||
) -> str:
|
||||
self._track_token_usage_internal(
|
||||
{"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110}
|
||||
)
|
||||
return "Thought: I know the answer.\nFinal Answer: fake answer"
|
||||
|
||||
def supports_function_calling(self) -> bool:
|
||||
return False
|
||||
|
||||
def supports_stop_words(self) -> bool:
|
||||
return False
|
||||
|
||||
def get_context_window_size(self) -> int:
|
||||
return 4096
|
||||
|
||||
|
||||
class TestUsageShapeUnitParity:
|
||||
def test_lite_agent_output_exposes_token_usage_object(self):
|
||||
metrics = UsageMetrics(
|
||||
total_tokens=110,
|
||||
prompt_tokens=100,
|
||||
completion_tokens=10,
|
||||
successful_requests=1,
|
||||
)
|
||||
output = LiteAgentOutput(
|
||||
agent_role="analyst", usage_metrics=metrics.model_dump()
|
||||
)
|
||||
|
||||
assert output.token_usage == metrics
|
||||
assert isinstance(output.token_usage, UsageMetrics)
|
||||
|
||||
def test_lite_agent_output_token_usage_zeroed_when_absent(self):
|
||||
output = LiteAgentOutput(agent_role="analyst")
|
||||
|
||||
assert output.usage_metrics is None
|
||||
assert output.token_usage == UsageMetrics()
|
||||
|
||||
def test_crew_output_exposes_usage_metrics_dict(self):
|
||||
metrics = UsageMetrics(
|
||||
total_tokens=110,
|
||||
prompt_tokens=100,
|
||||
completion_tokens=10,
|
||||
successful_requests=1,
|
||||
)
|
||||
output = CrewOutput(token_usage=metrics)
|
||||
|
||||
assert output.usage_metrics == metrics.model_dump()
|
||||
assert isinstance(output.usage_metrics, dict)
|
||||
|
||||
def test_both_shapes_carry_identical_keys(self):
|
||||
"""The dict shape has exactly the UsageMetrics fields on both types."""
|
||||
crew_dict = CrewOutput(token_usage=UsageMetrics()).usage_metrics
|
||||
lite = LiteAgentOutput(
|
||||
agent_role="analyst", usage_metrics=UsageMetrics().model_dump()
|
||||
)
|
||||
|
||||
assert set(crew_dict) == set(UsageMetrics.model_fields)
|
||||
assert set(lite.usage_metrics) == set(UsageMetrics.model_fields)
|
||||
|
||||
|
||||
class TestUsageShapeEndToEnd:
|
||||
"""Mirror of the EPD-178 clean-room repro, offline via a fake BaseLLM."""
|
||||
|
||||
@staticmethod
|
||||
def _read_via_object(result) -> int:
|
||||
"""Single accessor written against the CrewOutput shape."""
|
||||
return result.token_usage.prompt_tokens
|
||||
|
||||
@staticmethod
|
||||
def _read_via_dict(result) -> int:
|
||||
"""Single accessor written against the LiteAgentOutput shape."""
|
||||
return result.usage_metrics["prompt_tokens"]
|
||||
|
||||
def test_single_accessor_works_on_both_kickoff_paths(self):
|
||||
agent_a = Agent(
|
||||
role="analyst",
|
||||
goal="Answer questions.",
|
||||
backstory="Test agent.",
|
||||
llm=_FixedUsageLLM(),
|
||||
verbose=False,
|
||||
)
|
||||
result_agent = agent_a.kickoff("a question")
|
||||
|
||||
agent_b = Agent(
|
||||
role="analyst",
|
||||
goal="Answer questions.",
|
||||
backstory="Test agent.",
|
||||
llm=_FixedUsageLLM(),
|
||||
verbose=False,
|
||||
)
|
||||
task = Task(
|
||||
description="Answer: a question",
|
||||
expected_output="A short answer.",
|
||||
agent=agent_b,
|
||||
)
|
||||
crew = Crew(agents=[agent_b], tasks=[task], verbose=False)
|
||||
result_crew = crew.kickoff()
|
||||
|
||||
assert isinstance(result_agent, LiteAgentOutput)
|
||||
assert isinstance(result_crew, CrewOutput)
|
||||
|
||||
# Both accessors work on both result types and agree with each other.
|
||||
for result in (result_agent, result_crew):
|
||||
object_read = self._read_via_object(result)
|
||||
dict_read = self._read_via_dict(result)
|
||||
assert object_read == dict_read
|
||||
assert object_read > 0
|
||||
assert isinstance(result.token_usage, UsageMetrics)
|
||||
assert isinstance(result.usage_metrics, dict)
|
||||
@@ -18,11 +18,16 @@ def test_creating_a_tool_using_annotation():
|
||||
return question
|
||||
|
||||
assert my_tool.name == "Name of my tool"
|
||||
assert "Tool Name: name_of_my_tool" in my_tool.description
|
||||
assert "Tool Arguments:" in my_tool.description
|
||||
assert '"question"' in my_tool.description
|
||||
assert '"type": "string"' in my_tool.description
|
||||
assert "Tool Description: Clear description for what this tool is useful for" in my_tool.description
|
||||
# The authored description is preserved as written; the LLM-facing
|
||||
# composite lives at formatted_description.
|
||||
assert my_tool.description == (
|
||||
"Clear description for what this tool is useful for, your agent will need this information to use it."
|
||||
)
|
||||
assert "Tool Name: name_of_my_tool" in my_tool.formatted_description
|
||||
assert "Tool Arguments:" in my_tool.formatted_description
|
||||
assert '"question"' in my_tool.formatted_description
|
||||
assert '"type": "string"' in my_tool.formatted_description
|
||||
assert "Tool Description: Clear description for what this tool is useful for" in my_tool.formatted_description
|
||||
assert my_tool.args_schema.model_json_schema()["properties"] == {
|
||||
"question": {"title": "Question", "type": "string"}
|
||||
}
|
||||
@@ -33,9 +38,10 @@ def test_creating_a_tool_using_annotation():
|
||||
converted_tool = my_tool.to_structured_tool()
|
||||
assert converted_tool.name == "Name of my tool"
|
||||
|
||||
assert "Tool Name: name_of_my_tool" in converted_tool.description
|
||||
assert "Tool Arguments:" in converted_tool.description
|
||||
assert '"question"' in converted_tool.description
|
||||
assert converted_tool.description == my_tool.description
|
||||
assert "Tool Name: name_of_my_tool" in converted_tool.formatted_description
|
||||
assert "Tool Arguments:" in converted_tool.formatted_description
|
||||
assert '"question"' in converted_tool.formatted_description
|
||||
assert converted_tool.args_schema.model_json_schema()["properties"] == {
|
||||
"question": {"title": "Question", "type": "string"}
|
||||
}
|
||||
@@ -56,11 +62,16 @@ def test_creating_a_tool_using_baseclass():
|
||||
my_tool = MyCustomTool()
|
||||
assert my_tool.name == "Name of my tool"
|
||||
|
||||
assert "Tool Name: name_of_my_tool" in my_tool.description
|
||||
assert "Tool Arguments:" in my_tool.description
|
||||
assert '"question"' in my_tool.description
|
||||
assert '"type": "string"' in my_tool.description
|
||||
assert "Tool Description: Clear description for what this tool is useful for" in my_tool.description
|
||||
# The authored description is preserved as written; the LLM-facing
|
||||
# composite lives at formatted_description.
|
||||
assert my_tool.description == (
|
||||
"Clear description for what this tool is useful for, your agent will need this information to use it."
|
||||
)
|
||||
assert "Tool Name: name_of_my_tool" in my_tool.formatted_description
|
||||
assert "Tool Arguments:" in my_tool.formatted_description
|
||||
assert '"question"' in my_tool.formatted_description
|
||||
assert '"type": "string"' in my_tool.formatted_description
|
||||
assert "Tool Description: Clear description for what this tool is useful for" in my_tool.formatted_description
|
||||
assert my_tool.args_schema.model_json_schema()["properties"] == {
|
||||
"question": {"title": "Question", "type": "string"}
|
||||
}
|
||||
@@ -69,9 +80,10 @@ def test_creating_a_tool_using_baseclass():
|
||||
converted_tool = my_tool.to_structured_tool()
|
||||
assert converted_tool.name == "Name of my tool"
|
||||
|
||||
assert "Tool Name: name_of_my_tool" in converted_tool.description
|
||||
assert "Tool Arguments:" in converted_tool.description
|
||||
assert '"question"' in converted_tool.description
|
||||
assert converted_tool.description == my_tool.description
|
||||
assert "Tool Name: name_of_my_tool" in converted_tool.formatted_description
|
||||
assert "Tool Arguments:" in converted_tool.formatted_description
|
||||
assert '"question"' in converted_tool.formatted_description
|
||||
assert converted_tool.args_schema.model_json_schema()["properties"] == {
|
||||
"question": {"title": "Question", "type": "string"}
|
||||
}
|
||||
@@ -695,3 +707,88 @@ class TestToolDecoratorArunValidation:
|
||||
|
||||
with pytest.raises(ValueError, match="validation failed"):
|
||||
await async_execute.arun(wrong_arg="value")
|
||||
|
||||
|
||||
class TestAuthoredDescriptionPreserved:
|
||||
"""Regression tests for EPD-179: BaseTool.model_post_init silently
|
||||
rewrote the authored ``description`` into the LLM-facing composite
|
||||
(``Tool Name: …\\nTool Arguments: …\\nTool Description: <authored>``).
|
||||
The authored field must survive construction as written, with the
|
||||
composite exposed separately at ``formatted_description``.
|
||||
"""
|
||||
|
||||
AUTHORED = "Returns the current temperature for a city."
|
||||
|
||||
def _make_tool(self) -> BaseTool:
|
||||
class TempArgs(BaseModel):
|
||||
city: str = Field(description="City name to look up.")
|
||||
|
||||
class TempTool(BaseTool):
|
||||
name: str = "get_temperature"
|
||||
description: str = TestAuthoredDescriptionPreserved.AUTHORED
|
||||
args_schema: type[BaseModel] = TempArgs
|
||||
|
||||
def _run(self, city: str) -> str:
|
||||
return f"22C in {city}"
|
||||
|
||||
return TempTool()
|
||||
|
||||
def test_description_equals_authored_text(self):
|
||||
tool_instance = self._make_tool()
|
||||
assert tool_instance.description == self.AUTHORED
|
||||
|
||||
def test_formatted_description_contains_composite(self):
|
||||
tool_instance = self._make_tool()
|
||||
formatted = tool_instance.formatted_description
|
||||
assert "Tool Name: get_temperature" in formatted
|
||||
assert "Tool Arguments:" in formatted
|
||||
assert '"city"' in formatted
|
||||
assert formatted.endswith(f"Tool Description: {self.AUTHORED}")
|
||||
|
||||
def test_formatted_description_tracks_later_description_edits(self):
|
||||
tool_instance = self._make_tool()
|
||||
tool_instance.description = "Edited description."
|
||||
assert tool_instance.formatted_description.endswith(
|
||||
"Tool Description: Edited description."
|
||||
)
|
||||
|
||||
def test_prose_mentioning_the_marker_is_not_truncated(self):
|
||||
"""Authored text that merely mentions "Tool Description:" must reach
|
||||
the LLM untouched — only descriptions that ARE a pre-composed block
|
||||
(anchored three-line shape) get stripped."""
|
||||
tool_instance = self._make_tool()
|
||||
prose = (
|
||||
"Formats prompts. The output includes a line reading "
|
||||
"'Tool Description:' followed by the tool's summary."
|
||||
)
|
||||
tool_instance.description = prose
|
||||
assert tool_instance.formatted_description.endswith(
|
||||
f"Tool Description: {prose}"
|
||||
)
|
||||
|
||||
def test_composite_is_not_reapplied_to_prebaked_descriptions(self):
|
||||
"""A description that already contains a composed block (old
|
||||
checkpoints, adapters that bake the composite into the field) must
|
||||
not be double-wrapped."""
|
||||
tool_instance = self._make_tool()
|
||||
tool_instance.description = (
|
||||
"Tool Name: get_temperature\n"
|
||||
'Tool Arguments: {"city": "str"}\n'
|
||||
f"Tool Description: {self.AUTHORED}"
|
||||
)
|
||||
formatted = tool_instance.formatted_description
|
||||
assert formatted.count("Tool Description:") == 1
|
||||
assert formatted.endswith(f"Tool Description: {self.AUTHORED}")
|
||||
|
||||
def test_prompt_rendering_still_uses_composite(self):
|
||||
from crewai.utilities.agent_utils import render_text_description_and_args
|
||||
|
||||
tool_instance = self._make_tool()
|
||||
structured = tool_instance.to_structured_tool()
|
||||
assert structured.description == self.AUTHORED
|
||||
|
||||
for candidate in (tool_instance, structured):
|
||||
rendered = render_text_description_and_args([candidate])
|
||||
assert "Tool Name: get_temperature" in rendered
|
||||
assert "Tool Arguments:" in rendered
|
||||
assert f"Tool Description: {self.AUTHORED}" in rendered
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
"""Flow panels must be suppressed while a TUI owns the screen."""
|
||||
|
||||
from rich.text import Text
|
||||
|
||||
from crewai.events.listeners.tracing.utils import set_tui_mode
|
||||
from crewai.events.utils.console_formatter import ConsoleFormatter
|
||||
|
||||
|
||||
def _make_formatter(monkeypatch):
|
||||
fmt = ConsoleFormatter(verbose=True)
|
||||
calls: list[object] = []
|
||||
monkeypatch.setattr(fmt, "print", lambda *a, **k: calls.append(a))
|
||||
return fmt, calls
|
||||
|
||||
|
||||
def test_flow_panel_suppressed_in_tui_mode(monkeypatch):
|
||||
fmt, calls = _make_formatter(monkeypatch)
|
||||
set_tui_mode(True)
|
||||
try:
|
||||
fmt.print_panel(Text("x"), "🌊 Flow Started", "blue", is_flow=True)
|
||||
finally:
|
||||
set_tui_mode(False)
|
||||
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_flow_panel_prints_when_not_tui_mode(monkeypatch):
|
||||
fmt, calls = _make_formatter(monkeypatch)
|
||||
set_tui_mode(False)
|
||||
|
||||
fmt.print_panel(Text("x"), "🌊 Flow Started", "blue", is_flow=True)
|
||||
|
||||
# Panel + trailing blank line.
|
||||
assert len(calls) == 2
|
||||
|
||||
|
||||
def test_non_flow_panel_unaffected_by_tui_mode(monkeypatch):
|
||||
# tui_mode only gates flow panels; regular panels still follow verbose.
|
||||
fmt, calls = _make_formatter(monkeypatch)
|
||||
set_tui_mode(True)
|
||||
try:
|
||||
fmt.print_panel(Text("x"), "Task", "blue", is_flow=False)
|
||||
finally:
|
||||
set_tui_mode(False)
|
||||
|
||||
assert len(calls) == 2
|
||||
20
uv.lock
generated
20
uv.lock
generated
@@ -13,7 +13,7 @@ resolution-markers = [
|
||||
]
|
||||
|
||||
[options]
|
||||
exclude-newer = "2026-07-04T15:35:51.457693Z"
|
||||
exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values.
|
||||
exclude-newer-span = "P3D"
|
||||
|
||||
[options.exclude-newer-package]
|
||||
@@ -985,7 +985,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "chromadb"
|
||||
version = "1.1.1"
|
||||
version = "1.5.9"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "bcrypt" },
|
||||
@@ -1004,9 +1004,9 @@ dependencies = [
|
||||
{ name = "opentelemetry-sdk" },
|
||||
{ name = "orjson" },
|
||||
{ name = "overrides" },
|
||||
{ name = "posthog" },
|
||||
{ name = "pybase64" },
|
||||
{ name = "pydantic" },
|
||||
{ name = "pydantic-settings" },
|
||||
{ name = "pypika" },
|
||||
{ name = "pyyaml" },
|
||||
{ name = "rich" },
|
||||
@@ -1017,13 +1017,13 @@ dependencies = [
|
||||
{ name = "typing-extensions" },
|
||||
{ name = "uvicorn", extra = ["standard"] },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/7f/48/11851dddeadad6abe36ee071fedc99b5bdd2c324df3afa8cb952ae02798b/chromadb-1.1.1.tar.gz", hash = "sha256:ebfce0122753e306a76f1e291d4ddaebe5f01b5979b97ae0bc80b1d4024ff223", size = 1338109, upload-time = "2025-10-05T02:49:14.834Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/92/d1/5e33b26985f0c7046a0be1cee2158ada1748ee700d2545057fde1468d74d/chromadb-1.5.9.tar.gz", hash = "sha256:5c20e62a455c28bacac927f26116a73fd8e1799e0d908be8e8a4f02197a54731", size = 2595635, upload-time = "2026-05-05T05:54:51.713Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/39/59/0d881a9b7eb63d8d2446cf67fcbb53fb8ae34991759d2b6024a067e90a9a/chromadb-1.1.1-cp39-abi3-macosx_10_12_x86_64.whl", hash = "sha256:27fe0e25ef0f83fb09c30355ab084fe6f246808a7ea29e8c19e85cf45785b90d", size = 19175479, upload-time = "2025-10-05T02:49:12.525Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/94/4f/5a9fa317c84c98e70af48f74b00aa25589626c03a0428b4381b2095f3d73/chromadb-1.1.1-cp39-abi3-macosx_11_0_arm64.whl", hash = "sha256:95aed58869683f12e7dcbf68b039fe5f576dbe9d1b86b8f4d014c9d077ccafd2", size = 18267188, upload-time = "2025-10-05T02:49:09.236Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/45/1a/02defe2f1c8d1daedb084bbe85f5b6083510a3ba192ed57797a3649a4310/chromadb-1.1.1-cp39-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:06776dad41389a00e7d63d936c3a15c179d502becaf99f75745ee11b062c9b6a", size = 18855754, upload-time = "2025-10-05T02:49:03.299Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/5a/0d/80be82717e5dc19839af24558494811b6f2af2b261a8f21c51b872193b09/chromadb-1.1.1-cp39-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bba0096a7f5e975875ead23a91c0d41d977fbd3767f60d3305a011b0ace7afd3", size = 19893681, upload-time = "2025-10-05T02:49:06.481Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/2d/6e/956e62975305a4e31daf6114a73b3b0683a8f36f8d70b20aabd466770edb/chromadb-1.1.1-cp39-abi3-win_amd64.whl", hash = "sha256:a77aa026a73a18181fd89bbbdb86191c9a82fd42aa0b549ff18d8cae56394c8b", size = 19844042, upload-time = "2025-10-05T02:49:16.925Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/dd/5b/3cced915244f43ed14b53fe9f63a37f05f865064f4e4fe7d9448d3f2a352/chromadb-1.5.9-cp39-abi3-macosx_10_12_x86_64.whl", hash = "sha256:60701011b5e6409647fa40d12c7c5a66b2b0bfcf33a52db2ad53a30a2abc4957", size = 22564540, upload-time = "2026-05-05T05:54:48.906Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/34/4c/adcef1f4e82a2ef69ccd3711d55fc289193d54c4c0ff7a0292a3631db46f/chromadb-1.5.9-cp39-abi3-macosx_11_0_arm64.whl", hash = "sha256:814b9c95617377f6501e5757d63dfddb554a283a7739c87b9fa573850174e6f3", size = 21699698, upload-time = "2026-05-05T05:54:45.078Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/38/4e/937bc4d2e6f8ab9664ec79931fbbd69efff47e513ec2924b071e4b0ff774/chromadb-1.5.9-cp39-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:9192d111bd662241625867962333d99369a00769a50f8b2f58cb388731274d7e", size = 22680924, upload-time = "2026-05-05T05:54:36.25Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/e6/ec/0c42039e80b9acc534f67b73b7a42471948042859b3a64867b50a4a77fa3/chromadb-1.5.9-cp39-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:cc09b3df76e5a5cb386aed2715a2eea152e3949f9e1ba93c7119505377749929", size = 23316203, upload-time = "2026-05-05T05:54:41.157Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/eb/ce/0f7be6e5d0feafa2cda54b12e6542afeea7dea89d2d411e14da90f8abb96/chromadb-1.5.9-cp39-abi3-win_amd64.whl", hash = "sha256:4fd0b560e56761b7f3cb4d5c6205fd5f20814484b4a3e4e9af9038c2b428fc6c", size = 23542454, upload-time = "2026-05-05T05:54:54.942Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1430,7 +1430,7 @@ requires-dist = [
|
||||
{ name = "boto3", marker = "extra == 'aws'", specifier = "~=1.42.90" },
|
||||
{ name = "boto3", marker = "extra == 'bedrock'", specifier = "~=1.42.90" },
|
||||
{ name = "cel-python", specifier = ">=0.5.0,<0.6" },
|
||||
{ name = "chromadb", specifier = "~=1.1.0" },
|
||||
{ name = "chromadb", specifier = "~=1.5.9" },
|
||||
{ name = "click", specifier = ">=8.1.7,<9" },
|
||||
{ name = "crewai-cli", editable = "lib/cli" },
|
||||
{ name = "crewai-core", editable = "lib/crewai-core" },
|
||||
|
||||
Reference in New Issue
Block a user