د درس‌های عمیق‌تر فقط در نسخه‌های ابری

import cohere
co = cohere.Client(os.getenv("COHERE_API_KEY"))

# Step 1: over-retrieve from your vector DB
semantic_results = db.search(query, k=k * 10)
documents = [
    f"{r['metadata']['original_content']}\n\nContext: {r['metadata']['contextualized_content']}"
    for r in semantic_results
]

# Step 2: rerank
rerank_response = co.rerank(
    model="rerank-english-v3.0",
    query=query,
    documents=documents,
    top_n=k,
)

# Step 3: keep only the top-k reranked docs
final_docs = [
    {"chunk": semantic_results[r.index]["metadata"], "score": r.relevance_score}
    for r in rerank_response.results
]

<document>
{{WHOLE_DOCUMENT}}
</document>
Here is the chunk we want to situate within the whole document
<chunk>
{{CHUNK_CONTENT}}
</chunk>
Please give a short succinct context to situate this chunk within the
overall document for the purposes of improving search retrieval of the
chunk. Answer only with the succinct context and nothing else.

DOCUMENT_CONTEXT_PROMPT = """
<document>
{doc_content}
</document>
"""
CHUNK_CONTEXT_PROMPT = """
Here is the chunk we want to situate within the whole document
<chunk>
{chunk_content}
</chunk>

Please give a short succinct context to situate this chunk within the
overall document for the purposes of improving search retrieval of the
chunk. Answer only with the succinct context and nothing else.
"""

def situate_context(doc: str, chunk: str) -> str:
    response = client.messages.create(
        model="claude-haiku-4-5",   # cheap + fast for this task
        max_tokens=1024,
        temperature=0.0,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text",
                 "text": DOCUMENT_CONTEXT_PROMPT.format(doc_content=doc),
                 "cache_control": {"type": "ephemeral"}},
                {"type": "text",
                 "text": CHUNK_CONTEXT_PROMPT.format(chunk_content=chunk)},
            ],
        }],
    )
    return response.content[0].text

# At index time:
text_to_embed = f"{situate_context(doc, chunk)}\n\n{chunk}"
embedding = voyage_client.embed([text_to_embed], model="voyage-2").embeddings[0]

for chunk_id in chunk_ids:
    score = 0
    if chunk_id in ranked_chunk_ids:
        i = ranked_chunk_ids.index(chunk_id)
        score += semantic_weight * (1 / (i + 1))
    if chunk_id in ranked_bm25_chunk_ids:
        i = ranked_bm25_chunk_ids.index(chunk_id)
        score += bm25_weight * (1 / (i + 1))
    chunk_id_to_score[chunk_id] = score

batch_tool_schema = {
    "name": "batch_tool",
    "description": (
        "Invoke multiple other tools in parallel. Use whenever two or more "
        "independent tool calls are needed; they will all run concurrently."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "invocations": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "arguments": {"type": "string",
                                      "description": "JSON-encoded arguments"},
                    },
                    "required": ["name", "arguments"],
                },
            }
        },
        "required": ["invocations"],
    },
}

import asyncio, json

async def execute_batch(tool_use, registered_tools):
    invocations = tool_use["input"]["invocations"]
    coros = [
        registered_tools[inv["name"]](**json.loads(inv["arguments"]))
        for inv in invocations
    ]
    results = await asyncio.gather(*coros, return_exceptions=True)
    return [
        {"tool_name": inv["name"],
         "result": (str(r) if not isinstance(r, Exception) else f"ERROR: {r}")}
        for inv, r in zip(invocations, results)
    ]

# Inside the agent loop:
if tool_use["name"] == "batch_tool":
    batched_results = await execute_batch(tool_use, registered_tools)
    follow_up = client.messages.create(
        model=MODEL,
        max_tokens=2048,
        tools=all_tools + [batch_tool_schema],
        messages=history + [
            {"role": "assistant", "content": prev_response.content},
            {"role": "user", "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use["id"],
                "content": json.dumps(batched_results),
            }]},
        ],
    )

def auto_debug(failing_test, max_iters=5):
    history = [
        {"role": "user", "content":
         f"You are a debugging agent. The test below is failing.\n"
         f"Loop: (a) call run_tests, (b) read the failure, (c) propose ONE "
         f"minimal patch via str_replace_editor, (d) call run_tests again. "
         f"After each successful fix, call critique_self to argue why your "
         f"fix could still be wrong; iterate until critique_self returns "
         f"'no remaining objections'. Failing test: {failing_test}"},
    ]
    for _ in range(max_iters):
        resp = client.messages.create(model=MODEL, max_tokens=4096,
                                      tools=DEBUG_TOOLS, messages=history)
        history.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason == "end_turn": return history
        for block in resp.content:
            if block.type == "tool_use":
                tool_result = TOOLS[block.name](**block.input)
                history.append({"role": "user", "content": [{
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": tool_result,
                }]})
    return history